Credit: This is an IPython-ized version of the official AmpCamp 2013 Tutorial, all content credit goes to the AMPLab team.
Let's start by creating our Spark context:
from pyspark import SparkContext
sc = SparkContext( CLUSTER_URL, 'pyspark')
sc
<pyspark.context.SparkContext at 0x2dc4950>
1. Warm up by creating an RDD (Resilient Distributed Dataset) named pagecounts
from the input files, in this case 3 days of Wikipedia traffic:
pagecounts = sc.textFile("/wiki/pagecounts")
pagecounts
<pyspark.rdd.RDD at 0x3015d10>
2. Let’s take a peek at the data. You can use the take operation of an RDD to get the first K records. Here, K = 10.
pagecounts.take(10)
[u'20090505-000000 aa.b ?71G4Bo1cAdWyg 1 14463', u'20090505-000000 aa.b Special:Statistics 1 840', u'20090505-000000 aa.b Special:Whatlinkshere/MediaWiki:Returnto 1 1019', u'20090505-000000 aa.b Wikibooks:About 1 15719', u'20090505-000000 aa ?14mFX1ildVnBc 1 13205', u'20090505-000000 aa ?53A%2FuYP3FfnKM 1 13207', u'20090505-000000 aa ?93HqrnFc%2EiqRU 1 13199', u'20090505-000000 aa ?95iZ%2Fjuimv31g 1 13201', u'20090505-000000 aa File:Wikinews-logo.svg 1 8357', u'20090505-000000 aa Main_Page 2 9980']
3. Let’s see how many records in total are in this data set (this command will take a while, so read ahead while it is running).
%time pagecounts.count()
CPU times: user 64 ms, sys: 16 ms, total: 80 ms Wall time: 1min 46s
329641466
This should launch 177 Spark tasks on the Spark cluster. If you look closely at the terminal, the console log is pretty chatty and tells you the progress of the tasks. Because we are reading 20G of data from HDFS, this task is I/O bound and can take a while to scan through all the data (2 - 3 mins).  While it’s running, you can open the Spark web console to see the progress:
from IPython.display import HTML, IFrame, display
my_url = CLUSTER_URL.split(':')[1]
spark_console = 'http:'+my_url+':8080'
display(HTML('You can see the <a href="%s" target="_blank">Spark Console</a> here:'
% spark_console))
IFrame(spark_console, 800, 300)
4. Recall from above when we described the format of the data set, that the second field is the “project code” and contains information about the language of the pages. For example, the project code “en” indicates an English page. Let’s derive an RDD containing only English pages from pagecounts. This can be done by applying a filter function to pagecounts. For each record, we can split it by the field delimiter (i.e. a space) and get the second field-– and then compare it with the string “en”.
To avoid reading from disks each time we perform any operations on the RDD, we also cache the RDD into memory. This is where Spark really starts to to shine.
enPages = pagecounts.filter(lambda x: x.split(" ")[1] == "en").cache()
When you type this command into the Spark shell, Spark defines the RDD, but because of lazy evaluation, no computation is done yet. Next time any action is invoked on enPages, Spark will cache the data set in memory across the 5 slaves in your cluster.
5. How many records are there for English pages?
%time enPages.count()
CPU times: user 76 ms, sys: 36 ms, total: 112 ms Wall time: 2min 48s
122352588
The first time this command is run, similar to the last count we did, it will take 2 - 3 minutes while Spark scans through the entire data set on disk. But since enPages was marked as “cached” in the previous step, if you run count on the same RDD again, it should return an order of magnitude faster:
%time enPages.count()
CPU times: user 28 ms, sys: 32 ms, total: 60 ms Wall time: 6.26 s
122352588
If you examine the console log closely, you will see lines like this, indicating some data was added to the cache:
13/02/05 20:29:01 INFO storage.BlockManagerMasterActor$BlockManagerInfo: Added rdd_2_172 in memory on ip-10-188-18-127.ec2.internal:42068 (size: 271.8 MB, free: 5.5 GB)
6. Let’s try something fancier. Generate a histogram of total page views on Wikipedia English pages for the date range represented in our dataset (May 5 to May 7, 2009). The high level idea of what we’ll be doing is as follows. First, we generate a key value pair for each line; the key is the date (the first eight characters of the first field), and the value is the number of pageviews for that date (the fourth field).
enTuples = enPages.map(lambda x: x.split(" "))
enKeyValuePairs = enTuples.map(lambda x: (x[0][:8], int(x[3])))
Next, we shuffle the data and group all values of the same key together. Finally we sum up the values for each key. There is a convenient method called reduceByKey in Spark for exactly this pattern. Note that the second argument to reduceByKey determines the number of reducers to use. By default, Spark assumes that the reduce function is algebraic and applies combiners on the mapper side. Since we know there is a very limited number of keys in this case (because there are only 3 unique dates in our data set), let’s use only one reducer.
enKeyValuePairs.reduceByKey(lambda x, y: x + y, 1).collect()
[(u'20090506', 204190442), (u'20090507', 202617618), (u'20090505', 207698578)]
The collect method at the end converts the result from an RDD to an array.
We can combine the previous three commands into one:
enPages.map(
lambda x: x.split(" ")
).map(
lambda x: (x[0][:8], int(x[3]))
).reduceByKey(
lambda x, y: x + y, 1
).collect()
[(u'20090506', 204190442), (u'20090507', 202617618), (u'20090505', 207698578)]
7. Suppose we want to find pages that were viewed more than 200,000 times during the three days covered by our dataset. Conceptually, this task is similar to the previous query. But, given the large number of pages (23 million distinct page names), the new task is very expensive. We are doing an expensive group-by with a lot of network shuffling of data.
To recap, first we split each line of data into its respective fields. Next, we extract the fields for page name and number of page views. We reduce by key again, this time with 40 reducers. Then we filter out pages with less than 200,000 total views over our time window represented by our dataset.
%%time
fields = enPages.map(lambda x: x.split(" "))
title_nhits = fields.map(lambda x: (x[2], int(x[3])))
nhits = title_nhits.reduceByKey(lambda x, y: x + y, 40)
high_traffic = nhits.filter(lambda x: x[1] > 200000).map(lambda x: (x[1], x[0])).collect()
high_traffic
CPU times: user 104 ms, sys: 20 ms, total: 124 ms Wall time: 3min 27s
There is no hard and fast way to calculate the optimal number of reducers for a given problem; you will build up intuition over time by experimenting with different values.
You can explore the full RDD API by browsing the Java/Scala or Python API docs.
Because of time constraints, in this tutorial we focus on ad-hoc style analytics using the Spark shell. However, for many tasks, it makes more sense to write a standalone Spark program. We will return to this in the section on Spark Streaming below, where you will actually write a standalone Spark Streaming job. We aren’t going to cover how to structure, build, and run standalone Spark jobs here, but before we move on, we list here a few resources about standalone Spark jobs for you to come back and explore later.
First, on the AMI for this tutorial we have included “template” projects for Scala and Java standalone programs for both Spark and Spark streaming. The Spark ones can be found in the /root/scala-app-template
and /root/java-app-template
directories (we will discuss the Streaming ones later). Feel free to browse through the contents of those directories. You can also find examples of building and running Spark standalone jobs in Java and in Scala as part of the Spark Quick Start Guide. For even more details, see Matei Zaharia’s slides and talk video about Standalone Spark jobs at the first AMP Camp.