bigchungusdata/kafka/consumer.py
2021-04-25 13:03:26 +02:00

30 lines
712 B
Python

#!/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[strftime("%m/%d/%Y, %H:%M:%S")] = 1
else:
dt[strftime("%m/%d/%Y, %H:%M:%S")] +=1
for key, value in dt:
print(key,",",value)