python kafka producer

kafka_producer.py

import time
from kafka import KafkaProducer
import json
import base64

KAFKA_TOPIC = "scraped-data"
KAFKA_HOST = "localhost:9092"

producer = KafkaProducer(bootstrap_servers=KAFKA_HOST, value_serializer=lambda v: json.dumps(v).encode('utf-8'))

print "start producing to: " + KAFKA_TOPIC
localtime = time.localtime()
htime = time.strftime("%Y%m%d-%H%M%S", localtime)
print htime

for _ in range(3):
    rowkey = "msg_"+str(_)

    localtime = time.localtime()
    htime = time.strftime("%Y%m%d-%H%M%S", localtime)
    print htime

    data = {}
    data['rowkey'] = base64.b64encode(rowkey)
    data['col1'] = base64.b64encode('col_value_' + str(_))
    data['col2'] = base64.b64encode('col_value_' + str(_))
    data['col3'] = base64.b64encode('col_value_' + str(_))
    data['htime'] = base64.b64encode(htime)

    try:
        print data

        producer.send(KAFKA_TOPIC, data)

    except Exception as ex:
        print ex

producer.flush()

print "kafka-topic producer finished"

see http://kafka-python.readthedocs.io/en/master/usage.html
for more details