#!/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()