blob: 4e270a52e20ff94244268d4de682f45342073884 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
|
from kafka import KafkaConsumer
import json
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
import math
import string
import random
from pyspark.conf import SparkConf
topic_name = 'twitterdata'
# consumer = KafkaConsumer(topic_name)
# for msg in consumer:
# print (msg)
consumer = KafkaConsumer(
topic_name,
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 msg in consumer:
message = json.loads(json.dumps(msg.value))
try:
print(message['extended_tweet']['full_text'])
except:
try:
print(message['text'])
except:
print(message)
print(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>")
# spark = SparkSession.builder.master("local").appName("Tweets").getOrCreate()
# sc = spark.sparkContext
# sc.setLogLevel("ERROR")
# urlRdd = sc.parallelize([consumer])
# urldf = spark.read.json(urlRdd)
# urldf.printSchema()
# urldf.show()
# if __name__ == '__main__':
# spark = SparkSession.builder.appName("TwitterStreaming").getOrCreate().readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092").option("subscribe", topic_name).load()
# spark.createOrReplaceTempView("tweets")
# spark.sql("select * from tweets").show()
# spark.stop()
# sc = SparkContext(appName="TwitterStreaming")
# ssc = StreamingContext(sc, 5)
# ssc.checkpoint("checkpoint")
# lines = ssc.socketTextStream("localhost", 9999)
# words = lines.flatMap(lambda line: line.split(" "))
# pairs = words.map(lambda word: (word, 1))
# wordCounts = pairs.reduceByKey(lambda x, y: x + y)
# wordCounts.pprint()
# ssc.start()
# ssc.awaitTermination()
# ssc.stop()
# sc.stop()
# consumer.close()
# print("Done")
|
