===================TweetsRead.py===================
import tweepy
from tweepy import Stream
from tweepy.auth import OAuthHandler
from tweepy.streaming import StreamListener
import socket
import json
consumer_key = ''consumer_secret = ''access_token = ''access_secret = ''
class TweetsListener(StreamListener):
def __init__(self, csocket):
self.client_socket = csocket
def on_data(self, data):
try:
msg = json.loads(data)
print(msg['text'].encode('utf-8'))
self.client_socket.send(msg['text'].encode('utf-8'))
return True except BaseException as e:
print("Error on data: %s" % str(e))
return True
def on_error(self, status):
print(status)
return True
def sendData(c_socket):
auth = OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_secret)
twitter_stream = Stream(auth, TweetsListener(c_socket))
twitter_stream.filter(track=['china'])
if __name__ == "__main__":
s = socket.socket() # create socket object host = "127.0.0.1" port = 5555 s.bind((host, port))
print("Listening on port: %s" % str(port))
s.listen(5)
c, addr = s.accept()
print("Received request from: " + str(addr))
sendData(c)
===================TwitterAnalytics.py===================
import findspark
findspark.init('/Users/donghua/spark-2.3.2-bin-hadoop2.7')
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.sql import SQLContext
from pyspark.sql.functions import desc
# Not required if running in Pyspark integrated notebooksc = SparkContext()
ssc = StreamingContext(sc, 10)
sqlContext = SQLContext(sc)
socket_stream = ssc.socketTextStream("127.0.0.1",5555)
lines = socket_stream.window(20)
from collections import namedtuple
fields = ("tag","count")
Tweet = namedtuple('Tweet', fields)
# use () for multiple lines(lines.flatMap(lambda text: text.split(" "))
.filter(lambda word: word.startswith("#"))
.map(lambda word: (word.lower(), 1))
.reduceByKey(lambda a, b : a + b)
.map(lambda rec: Tweet(rec[0],rec[1]))
.foreachRDD(lambda rdd: rdd.toDF().sort(desc("count"))
.limit(10).registerTempTable("tweets")))
import time
from IPython import display
import matplotlib.pyplot as plt
import seaborn as sns
import pandas
# get_ipython().run_line_magic('matplotlib', 'inline')
ssc.start()
count = 0while count < 10:
time.sleep(10)
top_10_tweets = sqlContext.sql("select tag,count from tweets")
top_10_df = top_10_tweets.toPandas()
display.clear_output(wait = True)
plt.figure(figsize= (10, 8))
sns.barplot(x="count", y="tag",data=top_10_df)
plt.show()
count = count+1
ssc.stop()
