Consume JSON Messages From Kafka Using Kafka-Python’s Deserializer
Hope you are here when you want to take a ride on Python and Apache Kafka. Kafka-Python is most popular python library for Python. For documentation on this library visit to page https://kafka-python.readthedocs.io/en/master/. kafka-python is designed to function much like the official java client. kafka-python is best used with newer brokers (0.9+), but is backwards-compatible with older versions (to 0.8.0). Some features will only be enabled on newer brokers.
So instead of showing you a simple example to run Kafka Producer and Consumer separately, I’ll show the JSON serializer and deserializer.
Preparing the Environment
Lets start with Install python package using command below:-
pip install kafka-python
Import dependencies
import threading
import logging
import time
import jsonfrom kafka import KafkaConsumer, KafkaProducer
Create Class for Producer
You can send and receive strings if you remove the value_serializer and value_deserializer from the code below.
class Producer(threading.Thread):
daemon = True def run(self):
producer = KafkaProducer(bootstrap_servers='victoria.com:6667',
value_serializer=lambda v: json.dumps(v).encode('utf-8')) while True:
producer.send('my-topic', {"dataObjectID": "test1"})
producer.send('my-topic', {"dataObjectID": "test2"})
time.sleep(1)
Create Consumer Class
class Consumer(threading.Thread):
daemon = True def run(self):
consumer = KafkaConsumer(bootstrap_servers='victoria.com:6667',
auto_offset_reset='earliest',
value_deserializer=lambda m: json.loads(m.decode('utf-8')))
consumer.subscribe(['my-topic']) for message in consumer:
print (message)
Define Main thread: You can bundle above code into a single file and run on the console.
def main():
threads = [
Producer(),
Consumer()
] for t in threads:
t.start() time.sleep(10)if __name__ == "__main__":
logging.basicConfig(
format='%(asctime)s.%(msecs)s:%(name)s:%(thread)d:' +
'%(levelname)s:%(process)d:%(message)s',
level=logging.INFO
)
main()
Output: You can see the message has been shared between producer and consumer.
2017-11-19 02:18:30,590.590.334892273:kafka.client:140662616688384:INFO:20867:Bootstrap succeeded: found 1 brokers and 11 topics.
2017-11-19 02:18:30,694.694.561004639:kafka.conn:140662625081088:INFO:20867:Broker version identifed as 0.10.0
2017-11-19 02:18:30,694.694.842100143:kafka.conn:140662625081088:INFO:20867:Set configuration api_version=(0, 10, 0) to skip auto check_version requests on startup
2017-11-19 02:18:30,728.728.385925293:kafka.conn:140662616688384:INFO:20867:Broker version identifed as 0.10.0
2017-11-19 02:18:30,728.728.656053543:kafka.conn:140662616688384:INFO:20867:Set configuration api_version=(0, 10, 0) to skip auto check_version requests on startup
2017-11-19 02:18:30,730.730.935096741:kafka.coordinator.consumer:140662616688384:WARNING:20867:group_id is None: disabling auto-commit.
2017-11-19 02:18:30,731.731.933116913:kafka.consumer.subscription_state:140662616688384:INFO:20867:Updating subscribed topics to: ['my-topic']
2017-11-19 02:18:30,738.738.305091858:kafka.consumer.subscription_state:140662616688384:INFO:20867:Updated partition assignment: [TopicPartition(topic='my-topic', partition=0)]
ConsumerRecord(topic=u'my-topic', partition=0, offset=0, timestamp=1511075143446, timestamp_type=0, key=None, value={u'dataObjectID': u'test1'}, checksum=922149137, serialized_key_size=-1, serialized_value_size=25)
I am Getting this>>>>
ConsumerRecord(topic=u'my-topic', partition=0, offset=1, timestamp=1511075143447, timestamp_type=0, key=None, value={u'dataObjectID': u'test2'}, checksum=271715966, serialized_key_size=-1, serialized_value_size=25)
I am Getting this>>>>
ConsumerRecord(topic=u'my-topic', partition=0, offset=2, timestamp=1511075144449, timestamp_type=0, key=None, value={u'dataObjectID': u'test1'}, checksum=2115924298, serialized_key_size=-1, serialized_value_size=25)
I am Getting this>>>>
ConsumerRecord(topic=u'my-topic', partition=0, offset=3, timestamp=1511075144449, timestamp_type=0, key=None, value={u'dataObjectID': u'test2'}, checksum=2086194963, serialized_key_size=-1, serialized_value_size=25)
For complete code go to my github MyGit.