Spark-Reddit-App

reddit_spark.py

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
import json

def process_rdd(rdd):
    for record in rdd.collect():
        print(record)

sc = SparkContext('local[2]', 'RedditStreamApp')

ssc = StreamingContext(sc, 1)

TCP_IP = 'localhost'
TCP_PORT = 9010

lines = ssc.socketTextStream(TCP_IP, TCP_PORT)
lines.foreacRDD(lambda rdd: rdd.foreach(process_rdd))

ssc.start()
ssc.awaitTermination()

reddit_stream.py

from socket import socket, AF_INET, SOCK_STREAM
from praw import Reddit
import json

client_id = ''
client_secret = ''
user_agent = 'android:com.example.myredditapp:v1.2.3 (by u/maacpiash)'

reddit = Reddit(client_id=client_id, client_secret=client_secret, user_agent=user_agent)

TCP_IP = 'localhost'
TCP_PORT = 9010

s = socket(AF_INET, SOCK_STREAM)
s.bind((TCP_IP, TCP_PORT))
s.listen(1)

print(f'Listening on port: {TCP_PORT}')

conn, addr = s.accept()
subreddit = reddit.subreddit('all')

for submission in subreddit.stream.submissions():
    try:
        title = submission.title.lower()
        if 'happy' in title or 'money' in title:
            data = { 'title': submission.title, 'url': submission.url }
            print('Received a post:', json.dumps(data))
            conn.send(jsn.dumsp(data).encode('utf-8'))
    except Exception as e:
        print(e)