ml_kafka_create_consumer_producer_2.py

#1 — Initialize a Consumer for reading the email messages from the emails topic
consumer = KafkaConsumer(group_id="python-consumer",
                         bootstrap_servers=['localhost:9092'],
                         auto_offset_reset='earliest',# Always read from beginning
                         enable_auto_commit=False, # Keep Kafka messages as ‘unread’
                         value_deserializer=lambda x:json.loads(x))

consumer.subscribe("emails")

print(f'Initialized Kafka consumer at {dt.datetime.utcnow()})

#2 — Initialize a Producer for sending predictions to the predictions topic
producer = KafkaProducer(bootstrap_servers=['localhost:9092'],
                         value_serializer=lambda x:     
                         json.dumps(x,default=str).encode('utf-8'))

print(f'Initialized Kafka producer at {dt.datetime.utcnow()})