<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 <p class='note'>ReproduceIt is a series of articles that reproduce the results from data analysis articles focusing on having open data and open code.</p>
On this small article I reproduce the results (not visualization this time) from the reddit user /u/fhoffa, he posted a nice word cloud visualization of the most common words on some famous subreddits.
The reddit post can be found in the data is beautiful subreddit
Step1:  Note
Step2:  Create a data variable that points to the big 33 GB file and set the chunkbytes to 100 MB, map each item to json.loads since I know that each row is a JSON doc.
Step3:  Similar to spark you can take one or more items to see items of the dataset.
Step4:  Now some simple helper functions that we are going to use as filters.
Step5:  Now it comes the dask part which it will look really similar to anyone who has used Spark.
Step6:  Call compute and let dask work. At this moment if you take a look at your process manager you should a lot of Python processes.
Step7:  We can see how long the computation took, in this case
Step8:  With the word count done we can just sort the python list and see the most common words.
Step9:  Save a file with the results
<ASSISTANT_TASK:>
Python Code:
import re
import json
import time
import nltk
import dask
import dask.bag as db
import nltk
from nltk.corpus import stopwords
dask.__version__
Explanation: <p class='note'>ReproduceIt is a series of articles that reproduce the results from data analysis articles focusing on having open data and open code.</p>
On this small article I reproduce the results (not visualization this time) from the reddit user /u/fhoffa, he posted a nice word cloud visualization of the most common words on some famous subreddits.
The reddit post can be found in the data is beautiful subreddit: Reddit most common words for /r/politics, /r/movies, /r/trees, /r/science
and the original word cloud was this:
For some context he mentioned that he used Google BigQuery and Tableau. The data used was the most recent month available (May 2015) in a recent reddit dump that user /u/Stuck_In_the_Matrix made available in a nice torrent.
To reproduce the results I am using dask which is a nice new project from Continuum Analytics which got a lot of attention in the most recent SciPy. A little disclaimer here: I currently work for Continuum but this post is not sponsored in any way. 
Dask enables parallel computing through task scheduling and blocked algorithms. This allows developers to write complex parallel algorithms and execute them in parallel either on a modern multi-core machine or on a distributed cluster.
So basically dask is similar to Spark in the way of applying transformation to data but it also works amazingly well in a single multi-core machine and reuses battle tested libraries like numpy and pandas.
The data I am using is the same data as in the original post wich is 33.46 GB of a JSON new line delimited, each line having one comment with some metadata. All this code ran a my MacBook Pro (Retina, Mid 2014) - 2.8 GHz Intel core i7 - 16 GB 1600 MHz DDR3.
Lets jump to the code, as usual some libraries and their versions for reproducibility:
End of explanation
nltk.__version__
Explanation: Note: At the moment I am writing this you need the master version of dask since it includes a recent feature Matt Rocklin recently did but it should be available in the next version of dask.
Update: Dask 0.6.1 has been realeased so you no longer need to use master just conda install dask.
End of explanation
data = db.from_filenames("RC_2015-05", chunkbytes=100000).map(json.loads)
Explanation: Create a data variable that points to the big 33 GB file and set the chunkbytes to 100 MB, map each item to json.loads since I know that each row is a JSON doc.
End of explanation
data.take(1)
Explanation: Similar to spark you can take one or more items to see items of the dataset.
End of explanation
no_stopwords = lambda x: x not in stopwords.words('english')
is_word = lambda x: re.search("^[0-9a-zA-Z]+$", x) is not None
Explanation: Now some simple helper functions that we are going to use as filters.
End of explanation
subreddit = data.filter(lambda x: x['subreddit'] == 'movies')
bodies = subreddit.pluck('body')
words = bodies.map(nltk.word_tokenize).concat()
words2 = words.map(lambda x: x.lower())
words3 = words2.filter(no_stopwords)
words4 = words3.filter(is_word)
counts = words4.frequencies()
Explanation: Now it comes the dask part which it will look really similar to anyone who has used Spark.
End of explanation
start_time = time.time()
values = counts.compute()
elapsed_time = time.time() - start_time
Explanation: Call compute and let dask work. At this moment if you take a look at your process manager you should a lot of Python processes.
End of explanation
elapsed_time  # seconds
Explanation: We can see how long the computation took, in this case: around 23 minutes which is about 1.4 GB per minute.
End of explanation
len(values)
sort = sorted(values, key=lambda x: x[1], reverse=True)
sort[:100]
Explanation: With the word count done we can just sort the python list and see the most common words.
End of explanation
with open('r_movies.txt', 'w') as f:
    for item in sort:
        f.write("%s %i\n" % item)
Explanation: Save a file with the results
End of explanation
<END_TASK>