Added Kafka, Consumer and Producer Modules

1 files changed, 19 insertions, 0 deletions

diff --git a/consumer.py b/consumer.py
new file mode 100644
index 0000000..9b13a22
--- /dev/null
+++ b/consumer.py
@@ -0,0 +1,19 @@
+from kafka import KafkaConsumer
+import json
+
+topic_name = 'twitterdata'
+
+consumer = KafkaConsumer(
+    topic_name,
+    bootstrap_servers=['localhost:9092'],
+    auto_offset_reset='latest',
+    enable_auto_commit=True,
+    auto_commit_interval_ms =  5000,
+    fetch_max_bytes = 128,
+    max_poll_records = 100,
+
+    value_deserializer=lambda x: json.loads(x.decode('utf-8')))
+
+for message in consumer:
+    tweets = json.loads(json.dumps(message.value))
+    print(tweets)
\ No newline at end of file