Initial Commit

First Commit of the README, the lecture slides, the directroy structure,
and the first python files for Lecture 1.

Author: matthew-mcateer

README.md

@@ -0,0 +1,54 @@
+from pyspark import SparkConf,SparkContext
+from pyspark.streaming import StreamingContext
+from pyspark.sql import Row,SQLContext
+import sys
+import requests
+
+conf = SparkConf()
+conf.setAppName("TwitterStreamApp")
+sc = SparkContext(conf=conf)
+sc.setLogLevel("ERROR")
+ssc = StreamingContext(sc, 300)
+ssc.checkpoint("checkpoint_TwitterApp")
+dataStream = ssc.socketTextStream("localhost",9009)
+
+
+def aggregate_tags_count(new_values, total_sum):
+    return sum(new_values) + (total_sum or 0)
+
+
+def get_sql_context_instance(spark_context):
+    if ('sqlContextSingletonInstance' not in globals()):
+        globals()['sqlContextSingletonInstance'] = SQLContext(spark_context)
+    return globals()['sqlContextSingletonInstance']
+
+
+def send_df_to_dashboard(df):
+    top_tags = [str(t.hashtag) for t in df.select("hashtag").collect()]
+    tags_count = [p.hashtag_count for p in df.select("hashtag_count").collect()]
+    url = 'http://localhost:5001/updateData'
+    request_data = {'label': str(top_tags), 'data': str(tags_count)}
+    response = requests.post(url, data=request_data)
+
+
+def process_rdd(time, rdd):
+    print("----------- %s -----------" % str(time))
+    try:
+        sql_context = get_sql_context_instance(rdd.context)
+        row_rdd = rdd.map(lambda w: Row(hashtag=w[0], hashtag_count=w[1]))
+        hashtags_df = sql_context.createDataFrame(row_rdd)
+        hashtags_df.registerTempTable("hashtags")
+        hashtag_counts_df = sql_context.sql("select hashtag, hashtag_count from hashtags order by hashtag_count desc limit 5")
+        hashtag_counts_df.show()
+        send_df_to_dashboard(hashtag_counts_df)
+    except:
+        e = sys.exc_info()[0]
+        print("Error: %s" % e)
+
+words = dataStream.flatMap(lambda line: line.split(" "))
+hashtags = words.map(lambda x: (x, 1))
+tags_totals = hashtags.updateStateByKey(aggregate_tags_count)
+tags_totals.foreachRDD(process_rdd)
+
+ssc.start()
+ssc.awaitTermination()

section1/lectures/lecture1.pdf

@@ -0,0 +1,49 @@
+ACCESS_TOKEN = '2988835149-OgWXTLsBiosW74ZJi563l64WQ8f6tNudQKpLudp'
+ACCESS_SECRET = 'eOpeUX1wqL5sq9UK7yFbOPZ7ydYRZVqJ2Q2W7w3b1si7V'
+CONSUMER_KEY = 'AYYI7CvstplEi3fBAJ24vLVBA'
+CONSUMER_SECRET = 'OGFsmodX5DnHBcmZA3OrwFIeXS0gSUfPXcZVGTUOdffItb5Z0N'
+
+
+import socket
+import sys
+import requests
+import requests_oauthlib
+import json
+
+
+my_auth = requests_oauthlib.OAuth1(CONSUMER_KEY, CONSUMER_SECRET,ACCESS_TOKEN, ACCESS_SECRET)
+
+
+def send_tweets_to_spark(http_resp, tcp_connection):
+    for line in http_resp.iter_lines():
+        try:
+            full_tweet = json.loads(line)
+            tweet_text = full_tweet['text']
+            print("Tweet Text: " + tweet_text)
+            print ("------------------------------------------")
+            tcp_connection.send(tweet_text + '\n')
+        except:
+            e = sys.exc_info()[0]
+            print("Error: %s" % e)
+
+
+def get_tweets():
+    url = 'https://stream.twitter.com/1.1/statuses/filter.json'
+    query_data = [('locations', '-130,-20,100,50'), ('track', '#')]
+    query_url = url + '?' + '&'.join([str(t[0]) + '=' + str(t[1]) for t in query_data])
+    response = requests.get(query_url, auth=my_auth, stream=True)
+    print(query_url, response)
+    return response
+
+
+TCP_IP = "localhost"
+TCP_PORT = 9009
+conn = None
+s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+s.bind((TCP_IP, TCP_PORT))
+s.listen(1)
+print("Waiting for TCP connection...")
+conn, addr = s.accept()
+print("Connected... Starting getting tweets.")
+resp = get_tweets()
+send_tweets_to_spark(resp,conn)