Parsing Reddit Data with Dask

# Log into the EC2 instance created from a local terminal
ssh -i "~/.ssh/RandomKey.pem" ubuntu@ec2 -54-165-250-39.compute -1. amazonaws.com

# Upload the dataset to the instance from a local terminal   
scp -i ~/.ssh/RandomKey.pem ~/path_to_datafile/RC_2015 -01.bz2 ubuntu@ec2 -54-165-250-39.compute -1.amazonaws.com:~/data/RC_2015 -01. bz2

# Download Anaconda to the server from a server terminal
wget https://repo.continuum.io/archive/Anaconda2-4.2.0-Linux-x86_64. sh

# Install Anaconda to the instance from a server terminal
bash Anaconda2-4.2.0-Linux-x86_64.sh

# Open ipython in server environment and do the following command
ipython

# It will prompt a python interactive window that works in the bash environment
# Then create a password by typing the below commands
from IPython.lib import passwd passwd ()

# You will be prompted to enter and verify a password 
# (You should remember it. This will be the password to log to the remote ipython notebook)
# When done copy the generated sha1 key

# If you don’t already have a ipython_notebook_config.py file create one using the below command 
jupyter notebook --generate -config

# Write the below command to launch ipython
ipython notebook

# Load all libraries used
from bz2 import BZ2File
import ujson
import json
import re
import pandas as pd
from pandas import Timestamp, NaT, DataFrame
from toolz import dissoc
from castra import Castra
from toolz import peek, partition_all
import time
import datetime
import dask.dataframe as dd
import dask.array as da
from dask.diagnostics import ProgressBar
from dask.diagnostics import visualize
from dask.diagnostics import ResourceProfiler
import numpy as np
import nltk
from nltk.corpus import stopwords
import dask.bag as db
from bokeh.charts import TimeSeries, output_notebook, show
from string import punctuation
from itertools import izip
import matplotlib.pyplot as plt
import math
from wordcloud import WordCloud, STOPWORDS
from scipy.misc import imread

with BZ2File(’data/RC_2015-01.bz2’) as f: 
    line = f.readline()
ujson.loads(line)

# Create a path for locating the data files
path = "data"
raw_file = "RC_2015-01.bz2"

# Give a number of lines processed per attempt
# Use a small number because of the small memory size(1GB) of the instance
lppa = 20000

#Convert a line of JSON(post) into a cleaned up dict.
def to_json(line):
    post = ujson.loads(line)

    # Convert timestamps into Timestamp objects.
    # It will help us create the sentiment analysis over time date = post["created_utc"]
    post["created_utc"] = Timestamp.utcfromtimestamp(int(date)) edited = post["edited"]
    post["edited"] = Timestamp.utcfromtimestamp(int(edited)) if edited else NaT

    # Convert deleted posts into None values
    if post["author"] == "[deleted]": 
        post["author"] = None
    if post["body"] == "[deleted]":
        post["body"] = None

    # Remove "id", and "subreddit_id" as they’re redundant
    # Remove "retrieved_on" as it’s irrelevant
    return dissoc(post, "id", "subreddit_id", "retrieved_on")

# Store the column names for each post in a variable
columns = ["archived", "author", "author_flair_css_class", "author_flair_text", "body", "controversiality", "created_utc", "distinguished", "downs", "edited", "gilded", "link_id", "name", " parent_id", "removal_reason", "score", "score_hidden", "subreddit" , "ups"]

# Convert a list of json strings into a dataframe
def to_df(batch):
    posts = map(to_json, batch)
    df = DataFrame.from_records(posts, columns=columns)
    return df.set_index("created_utc")

#Creating the castra file
print "Processing.."
start = time.time()
categories = ["distinguished", "subreddit", "removal_reason"] 
with BZ2File(path+raw_file) as f:
    batches = partition_all(lppa, f)
    df, frames = peek(map(to_df, batches))
    castra = Castra("reddit_data.castra", template=df, categories=categories)
    castra.extend_sequence(frames, freq="3h")

end = time.time()
print "Castra file has been created in %s:" % str(datetime.timedelta(seconds=(end-start)))

# Create the castra dataset. This will help with I/O operations 
'''
def to_castra(fullpath,lppa):
    categories = [’distinguished’, ’subreddit’, ’removal_reason’] 
    filename = fullpath.split(’/’)[-1].split(’.’)[0]
    with BZ2File(fullpath) as f:
        batches = partition_all(lppa, f)
        castra = None
        for batch in batches:
            df = to_df(batch) 
            if castra == None:
                castra = Castra(path+’reddit_’+filename+’.castra’, template=df, categories=categories)
            castra.extend(df)
'''

'''
print "Processing.."

#Creating the castra file 
start = time.time() 
to_castra(path+raw_file,lppa) 
end = time.time()
print "Castra file has been created in %s:" % str(datetime.timedelta(seconds=(end-start)))
'''

# Start ram and CPU visualizer for all computations
a = da.random.random(size=(10000, 1000), chunks=(1000, 1000))
q, r = da.linalg.qr(a)
a2 = q.dot(r)

with ResourceProfiler(dt=0.25) as rprof:
    out = a2.compute()
rprof.visualize()

# Start a progress bar for all computations
pbar = ProgressBar()
pbar.register()

# Load data into a dask dataframe:
path_castra = "data/reddit_RC_2015-01.castra"
df = dd.from_castra(path_castra, columns=columns)
df.head(5)

# General info about the dataset
df.info()

# The total number of posts
df.downs.count().compute()

# Top 10 up-rated values
df.ups.nlargest(10).compute()

# Cell counts for the top 10 subreddits
df.groupby("subreddit").count().nlargest(10).compute()

# The posts with ups>5000
df[df[’ups’]>5000].compute()

# Create a Series with number of counts per 12 hours
counts_12h = df.ups.resample(’12h’).count().compute()

# Create a lineplot
plt.figure(num=1,figsize=(12, 8), dpi=80)
line, = plt.plot(counts_12h.index, counts_12h.values, lw=2) 
plt.axis()
plt.xlabel("Date")
plt.ylabel("Posts") plt.title("All Comments") 
plt.show()

# Load first 5 lines of file
columns = ["word", "happiness_rank", "happiness_avg", "happiness_std" , "twitter_rank", "google_rank", "nyt_rank", "lyrics_rank"]
happy_file = pd.read_csv("data/Data_Set_S1.txt", sep="\t", names= columns, encoding="latin-1")[1:]
happy_file.head()

# Change data type to numeric
happy_file[["happiness_avg"]]= happy_file[["happiness_avg"]].apply( pd.to_numeric)

# Happiness average value range
print "happiness value range : { %f - %f } \nmean \t\t\t: { %f }" % \ 
(happy_file.happiness_avg.min(),happy_file.happiness_avg.max(), 
happy_file.happiness_avg.mean())

# Create happy_dictionary
happy_dict = {} 
index = 1
for k in happy_file["word"]:
    happy_dict[k] = happy_file["happiness_avg"][index] 
    index +=1

# Load castra file into a dask bag
bag = db.from_castra('reddit_data.castra/', columns=['subreddit', 'body'])

# Show a sample of the first post
bag.take(1)

# Save files with subreddit "worldnews"
matches_subreddit = bag.filter(lambda x: x["subreddit"] == "worldnews")

# Collect only the bodies of posts
bodies = matches_subreddit.pluck("body")

# Create a bag of tokens
dirty_bag = bodies.map(nltk.word_tokenize).concat()

# Lower all characters in the bag
lower_bag = dirty_bag.map(lambda x: x.lower())

# Clean bag from stopwords
clean_bag = lower_bag.filter(lambda x: x not in stopwords.words("english"))

# keep only alphanumeric characters as words in the bag and remove NoneType
bag_of_words = clean_bag.filter(lambda x: re.search("^[0-9a-zA-Z]+$", x) is not None)

print "Processing.."
start = time.time()

# Keep all of the above in a variable in memory by using .compute()
bag_of_words = bag_of_words.compute()

end = time.time()
print "The bag of words has been created in: %s" % str(datetime.timedelta(seconds=(end-start)))

# Save bag_of_words to a file to release cache memory usage
with open("data/bag_of_words.csv", "w") as f:
    for item in bag_of_words: 
        f.write("%s\n" % item)

# Restart kernel and load the file again
with open("data/bag_of_words.csv", "r") as f:
    bag_of_words = [item[:-1] for item in f]

# Print no. of words in the bag
print 'No. of words in the bag: ', len(bag_of_words)

def happy_meter(tokens):
    # Check if tokens match happy_dict counter = 0
    total_ratio = 0
    for i in happy_dict.keys():
        for token in tokens:
            if token in i:
                counter+=1
                total_ratio+=float(happy_dict[i])

    # print "Happiness ratio range is (1.30 - 8.50)"
    if counter == 0:
        return 0 #"There are no words with sentiment attached"
    else:
        return total_ratio/counter
        #"The happiness ratio of this text is: {0:.2f}".format(total_ratio/counter)

def sentimentizer(bag,n):
    
    # Create buckets of words
    buckets = izip(*[iter(bag)]*n)

    # Calculate sentiment of each bucket and assign values to new dictionary
    counter = 1
    ratio = [] 
    word_counter = []
    for bucket in buckets:
        ratio.append(happy_meter(list(bucket))) 
        word_counter.append(n*counter) 
        counter+=1

    # Create a lineplot for sentiment profile
    plt.figure(num=1,figsize=(12, 8), dpi=80) 
    line, = plt.plot(count_12h.index, ratio, lw=2) 
    plt.axis()
    plt.xlabel(’Date’)
    plt.ylabel(’Happiness Ratio’) plt.title(’Sentiment Profile’)
    plt.show()

# Create profiles for sliding windows of length 500 words
# It takes almost 4 hours to run!!
sentimentizer(bag_of_words,500)

# Create a function that turns the bag_of_words into a string
def generate_string(bag):
    s = ''
    s+=' '.join([word for word in bag])
    return s

# Generate string from bag_of_words
s = generate_string(bag_of_words)

# Exclude some more from the cloud
stopwords = {"deleted", "gt"}
STOPWORDS = STOPWORDS.union(stopwords)

# Create a mask as a stencil for the word cloud
world_mask = imread("photos/world_3.jpg",flatten=True)

# Set requirements for the bag_of_words
wc = WordCloud(background_color="white", max_words=2000, width=1800, height=1400, mask=world_mask, stopwords=STOPWORDS)

# Generate the word cloud by sourcing the string
wc.generate(s)

#  Save the word cloud
#plt.imshow(wc)
plt.axis("off")
plt.savefig("photos/world_cloud.png", dpi=300)
plt.figure(figsize=(12,8),dpi=500)
img=plt.imshow(wc)
#plt.imshow(world_mask, cmap=plt.cm.gray)
plt.axis("off")
plt.show()