bigchungusdata/kafka/consumer.py

#!/usr/bin/env python3
from datetime import datetime

from kafka import KafkaConsumer


DATETIMES=[]


consumer = KafkaConsumer(bootstrap_servers='localhost:9092', auto_offset_reset='earliest', consumer_timeout_ms=1000)
consumer.subscribe(['test'])


try:
    while True:
        for message in consumer:
            DATETIMES.append(datetime.now())
    consumer.close()
except KeyboardInterrupt:
    consumer.close()
    dt = {}
    for datetime in DATETIMES:
        dts = datetime.strftime("%m/%d/%Y, %H:%M:%S")
        if dts not in dt:
            dt[dts] = 1
        else:
            dt[dts] +=1
    for key in dt:
        print(key,",",dt[key])
    consumer.close()