Using Glue schema registry for Apache Kafka with Python

This video explains a Python streaming data pipeline which leverage schemas for data validation using Kafka with AVRO and AWS Glue Schema Registry.

Documentation Links:
-----------------------------------
https://pypi.org/project/aws-glue-schema-registry/
https://docs.aws.amazon.com/glue/latest/dg/schema-registry.html

Prerequisites:
---------------------------
Introduction to Schema Registry in Kafka | Part 1
https://youtu.be/nCiTat06Ius
Introduction to Schema Registry in Kafka | Part 2
https://youtu.be/46aZ6dqvswk

Avro Schema Used:
------------------------------------
{
"namespace": "aws_schema_registry.integrationtests",
"type": "record",
"name": "User",
"fields": [
{"name": "name", "type": "string"},
{"name": "Age", "type": "int"}
]
}

Python Code:
--------------------------
#pip3 install boto3 -t.
#pip3 install aws-glue-schema-registry --upgrade --use-pep517 -t .
#pip install kafka-python -t .
import boto3
from time import sleep
from json import dumps
from kafka import KafkaProducer
from aws_schema_registry import DataAndSchema, SchemaRegistryClient
from aws_schema_registry.avro import AvroSchema
from aws_schema_registry.adapter.kafka import KafkaSerializer

session = boto3.Session(aws_access_key_id='{}', aws_secret_access_key='{}')

glue_client = session.client('glue', region_name='us-east-1')


# Create the schema registry client, which is a façade around the boto3 glue client
client = SchemaRegistryClient(glue_client,
registry_name='my-registry')

# Create the serializer
serializer = KafkaSerializer(client)

# Create the producer
producer = KafkaProducer(bootstrap_servers=['127.0.0.1:9092'],value_serializer=serializer)


# Our producer needs a schema to send along with the data.
# In this example we're using Avro, so we'll load an .avsc file.
with open('C:/Users/USER/PycharmProjects/kafka_python/user.avsc', 'r') as schema_file:
schema = AvroSchema(schema_file.read())



# Send message data along with schema
data = {
'name': 'Hello',
'Age':45
}
#data={'Partiiton_no':2}
record_metadata =producer.send('glue_schema_bms', value=(data, schema)).get(timeout=10)
print(record_metadata.topic)
print(record_metadata.partition)
print(record_metadata.offset)

Check this playlist for more Data Engineering related videos:
https://youtube.com/playlist?list=PLjfRmoYoxpNopPjdACgS5XTfdjyBcuGku

Apache Kafka form scratch
https://youtube.com/playlist?list=PLjfRmoYoxpNrs0VmIq6mOTqXP52RfZdRf

Snowflake Complete Course from scratch with End-to-End Project with in-depth explanation--
https://doc.clickup.com/37466271/d/h/13qc4z-104/d4346819bd8d510

????????
YOU JUST NEED TO DO
3 THINGS to support my channel
LIKE
SHARE
&
SUBSCRIBE
TO MY YOUTUBE CHANNEL