CODEWITHSUNDEEP
pythonapache-kafkaavro
korador
korador

Reputation: 5

Kafka + Avro + Python

I want to send messages by producer and getting them by consumer. It has to be in avro, but i dont know how to do it. Take a look:

schema = { "type":"record", "name":"myrecord", "fields": [{"name":"typ","type":"string"}, {"name":"pred","type":"int"}

producer = KafkaProducer(bootstrap_servers=['xxxx:xxxx'],value_serializer = avro.schema.parse(json.dumps(schema)))

for i in range(100):
    message = {"typ":"sth","pred":i}
    producer.send("xxxx", value=message)

Can you help me how to do it correctly?

Upvotes: 0

Views: 6627

Answers (2)

dalelane
dalelane

Reputation: 2765

Something like this should do the trick:

from kafka import KafkaProducer
import io
from avro.schema import Parse
from avro.io import DatumWriter, DatumReader, BinaryEncoder, BinaryDecoder

# Create a Kafka client ready to produce messages
producer = KafkaProducer(bootstrap_servers=bootstrap_address,
                         security_protocol="...", ...)

# Get the schema to use to serialize the message
schema = Parse(open(FILENAME_WHERE_YOU_HAVE_YOUR_AVRO_SCHEMA, "rb").read())

# serialize the message data using the schema
buf = io.BytesIO()
encoder = BinaryEncoder(buf)
writer = DatumWriter(writer_schema=schema)
writer.write(myobject, encoder)
buf.seek(0)
message_data = (buf.read())

# message key if needed
key = None

# headers if needed
headers = []

# Send the serialized message to the Kafka topic
producer.send(topicname,
              message_data,
              key,
              headers)
producer.flush()

Upvotes: 4

OneCricketeer
OneCricketeer

Reputation: 191743

Using kafka-python, the value_serializer needs to be a function of the value, not a parsed Avro schema.

For example

from avro.io import DatumWriter

schema_def = { ... }
schema = avro.schema.parse(json.dumps(schema_def).encode('utf-8'))

def serialize(value):
    writer = DatumWriter()
    # TODO: add schema to writer
    # TODO: write value payload to writer
    # TODO: return writer to bytes

producer = KafkaProducer(value_serializer=serialize)

This is more work than you really need to do. Look at the confluent-kafka-python example code instead - https://github.com/confluentinc/confluent-kafka-python/blob/master/examples/avro_producer.py

Upvotes: 0

Related Questions