# Fisseha Berhane, PhD

#### Data Scientist

443-970-2353 [email protected] CV Resume

# Building a word count application¶

### Part 1: Creating a base RDD and pair RDDs ¶

#### We'll start by generating a base RDD by using a Python list and the sc.parallelize method. Then we'll print out the type of the base RDD.¶

In [38]:
wordsList = ['cat', 'elephant', 'rat', 'rat', 'cat']
wordsRDD = sc.parallelize(wordsList, 4)
# Print out the type of wordsRDD
print type(wordsRDD)

<class 'pyspark.rdd.RDD'>


#### This is the general form that exercises will take, except that no example solution will be provided. Exercises will include an explanation of what is expected, followed by code cells where one cell will have one or more <FILL IN> sections. The cell that needs to be modified will have # TODO: Replace <FILL IN> with appropriate code on its first line. Once the <FILL IN> sections are updated and the code is run, the test cell can then be run to verify the correctness of your solution. The last code cell before the next markdown section will contain the tests.¶

In [39]:
# TODO: Replace <FILL IN> with appropriate code
def makePlural(word):
"""Adds an 's' to word.

Note:
This is a simple function that only adds an 's'.  No attempt is made to follow proper
pluralization rules.

Args:
word (str): A string.

Returns:
str: A string with 's' added to it.
"""
return (word+'s')

print makePlural('cat')

cats

In [40]:
# One way of completing the function
def makePlural(word):
return word + 's'

print makePlural('cat')

cats

In [41]:
# Load in the testing code and check to see if your answer is correct
# If incorrect it will report back '1 test failed' for each failed test
# Make sure to rerun any cell you change before trying the test again
from test_helper import Test
# TEST Pluralize and test (1b)
Test.assertEquals(makePlural('rat'), 'rats', 'incorrect result: makePlural does not add an s')

1 test passed.


#### Now pass each item in the base RDD into a map() transformation that applies the makePlural() function to each element. And then call the collect() action to see the transformed RDD.¶

In [42]:
# TODO: Replace <FILL IN> with appropriate code
pluralRDD = wordsRDD.map(makePlural)
print pluralRDD.collect()

['cats', 'elephants', 'rats', 'rats', 'cats']

In [43]:
# TEST Apply makePlural to the base RDD(1c)
Test.assertEquals(pluralRDD.collect(), ['cats', 'elephants', 'rats', 'rats', 'cats'],
'incorrect values for pluralRDD')

1 test passed.


#### Let's create the same RDD using a lambda function.¶

In [44]:
# TODO: Replace <FILL IN> with appropriate code
pluralLambdaRDD = wordsRDD.map(lambda x:x+'s')
print pluralLambdaRDD.collect()

['cats', 'elephants', 'rats', 'rats', 'cats']

In [45]:
# TEST Pass a lambda function to map (1d)
Test.assertEquals(pluralLambdaRDD.collect(), ['cats', 'elephants', 'rats', 'rats', 'cats'],
'incorrect values for pluralLambdaRDD (1d)')

1 test passed.


#### Now use map() and a lambda function to return the number of characters in each word. We'll collect this result directly into a variable.¶

In [46]:
# TODO: Replace <FILL IN> with appropriate code
pluralLengths = (pluralRDD
.map(lambda x: len(x))
.collect())
print pluralLengths

[4, 9, 4, 4, 4]

In [47]:
# TEST Length of each word (1e)
Test.assertEquals(pluralLengths, [4, 9, 4, 4, 4],
'incorrect values for pluralLengths')

1 test passed.


#### We can create the pair RDD using the map() transformation with a lambda() function to create a new RDD.¶

In [48]:
# TODO: Replace <FILL IN> with appropriate code
wordPairs = wordsRDD.map(lambda x: (x,1))
print wordPairs.collect()

[('cat', 1), ('elephant', 1), ('rat', 1), ('rat', 1), ('cat', 1)]

In [49]:
# TEST Pair RDDs (1f)
Test.assertEquals(wordPairs.collect(),
[('cat', 1), ('elephant', 1), ('rat', 1), ('rat', 1), ('cat', 1)],
'incorrect value for wordPairs')

1 test passed.


### Part 2: Counting with pair RDDs ¶

#### An approach you might first consider (we'll see shortly that there are better ways) is based on using the groupByKey() transformation. As the name implies, the groupByKey() transformation groups all the elements of the RDD with the same key into a single list in one of the partitions. There are two problems with using groupByKey():¶

• #### The operation requires a lot of data movement to move all the values into the appropriate partitions.
• #### The lists can be very large. Consider a word count of English Wikipedia: the lists for common words (e.g., the, a, etc.) would be huge and could exhaust the available memory in a worker.

#### Use groupByKey() to generate a pair RDD of type ('word', iterator).¶

In [50]:
# TODO: Replace <FILL IN> with appropriate code
# Note that groupByKey requires no parameters
wordsGrouped = wordPairs.groupByKey().mapValues(lambda x: list(x))
for key, value in wordsGrouped.collect():
print '{0}: {1}'.format(key, list(value))

rat: [1, 1]
elephant: [1]
cat: [1, 1]

In [51]:
# TEST groupByKey() approach (2a)
Test.assertEquals(sorted(wordsGrouped.mapValues(lambda x: list(x)).collect()),
[('cat', [1, 1]), ('elephant', [1]), ('rat', [1, 1])],
'incorrect value for wordsGrouped')

1 test passed.


#### Now sum the iterator using a map() transformation. The result should be a pair RDD consisting of (word, count) pairs.¶

In [52]:
# TODO: Replace <FILL IN> with appropriate code
wordCountsGrouped = wordsGrouped.map(lambda (k, v): (k, sum(v)))
print wordCountsGrouped.collect()

[('rat', 2), ('elephant', 1), ('cat', 2)]

In [53]:
# TEST Use groupByKey() to obtain the counts (2b)
Test.assertEquals(sorted(wordCountsGrouped.collect()),
[('cat', 2), ('elephant', 1), ('rat', 2)],
'incorrect value for wordCountsGrouped')

1 test passed.


#### A better approach is to start from the pair RDD and then use the reduceByKey() transformation to create a new pair RDD. The reduceByKey() transformation gathers together pairs that have the same key and applies the function provided to two values at a time, iteratively reducing all of the values to a single value. reduceByKey() operates by applying the function first within each partition on a per-key basis and then across the partitions, allowing it to scale efficiently to large datasets.¶

In [54]:
# TODO: Replace <FILL IN> with appropriate code
# Note that reduceByKey takes in a function that accepts two values and returns a single value
wordCounts = wordPairs.reduceByKey(lambda a,b: a+b)
print wordCounts.collect()

[('rat', 2), ('elephant', 1), ('cat', 2)]

In [55]:
# TEST Counting using reduceByKey (2c)
Test.assertEquals(sorted(wordCounts.collect()), [('cat', 2), ('elephant', 1), ('rat', 2)],
'incorrect value for wordCounts')

1 test passed.


#### The expert version of the code performs the map() to pair RDD, reduceByKey() transformation, and collect in one statement.¶

In [56]:
# TODO: Replace <FILL IN> with appropriate code
wordCountsCollected =(wordsRDD.map(lambda x: (x,1)).reduceByKey(lambda a,b:a+b)
.collect())
print wordCountsCollected

[('rat', 2), ('elephant', 1), ('cat', 2)]

In [57]:
# TEST All together (2d)
Test.assertEquals(sorted(wordCountsCollected), [('cat', 2), ('elephant', 1), ('rat', 2)],
'incorrect value for wordCountsCollected')

1 test passed.


### Part 3: Finding unique words and a mean value ¶

#### Calculate the number of unique words in wordsRDD. You can use other RDDs that you have already created to make this easier.¶

In [58]:
# TODO: Replace <FILL IN> with appropriate code
uniqueWords = len(wordCountsCollected)
print uniqueWords

3

In [59]:
# TEST Unique words (3a)
Test.assertEquals(uniqueWords, 3, 'incorrect count of uniqueWords')

1 test passed.


#### Use a reduce() action to sum the counts in wordCounts and then divide by the number of unique words. First map() the pair RDD wordCounts, which consists of (key, value) pairs, to an RDD of values.¶

In [60]:
# TODO: Replace <FILL IN> with appropriate code
totalCount = (wordCounts
.map(lambda (x,v): v)
average = totalCount / float(uniqueWords)
print totalCount
print round(average, 2)

5
1.67

In [61]:
# TEST Mean using reduce (3b)
Test.assertEquals(round(average, 2), 1.67, 'incorrect value of average')

1 test passed.


### Part 4: Apply word count to a file ¶

#### First, define a function for word counting. You should reuse the techniques that have been covered in earlier parts of this lab. This function should take in an RDD that is a list of words like wordsRDD and return a pair RDD that has all of the words and their associated counts.¶

In [62]:
# TODO: Replace <FILL IN> with appropriate code
def wordCount(wordListRDD):
"""Creates a pair RDD with word counts from an RDD of words.

Args:
wordListRDD (RDD of str): An RDD consisting of words.

Returns:
RDD of (str, int): An RDD consisting of (word, count) tuples.
"""
return (wordListRDD.map(lambda x: (x,1)).reduceByKey(lambda a,b:a+b))
print wordCount(wordsRDD).collect()

[('rat', 2), ('elephant', 1), ('cat', 2)]

In [63]:
# TEST wordCount function (4a)
Test.assertEquals(sorted(wordCount(wordsRDD).collect()),
[('cat', 2), ('elephant', 1), ('rat', 2)],
'incorrect definition for wordCount function')

1 test passed.


#### Real world files are more complicated than the data we have been using in this lab. Some of the issues we have to address are:¶

• #### Words should be counted independent of their capitialization (e.g., Spark and spark should be counted as the same word).
• #### All punctuation should be removed.
• #### Any leading or trailing spaces on a line should be removed.

#### Define the function removePunctuation that converts all text to lower case, removes any punctuation, and removes leading and trailing spaces. Use the Python re module to remove any text that is not a letter, number, or space. Reading help(re.sub) might be useful.¶

In [64]:
# TODO: Replace <FILL IN> with appropriate code
import re
def removePunctuation(text):
"""Removes punctuation, changes to lower case, and strips leading and trailing spaces.

Note:
Only spaces, letters, and numbers should be retained.  Other characters should should be
eliminated (e.g. it's becomes its).  Leading and trailing spaces should be removed after
punctuation is removed.

Args:
text (str): A string.

Returns:
str: The cleaned up string.
"""
return re.sub(r'[^a-zA-Z0-9\s]+','',text).lower().strip()
print removePunctuation('Hi, you!')
print removePunctuation(' No under_score!')

hi you
no underscore

In [65]:
# TEST Capitalization and punctuation (4b)
Test.assertEquals(removePunctuation(" The Elephant's 4 cats. "),
'the elephants 4 cats',
'incorrect definition for removePunctuation function')

1 test passed.


#### For the next part of this lab, we will use the Complete Works of William Shakespeare from Project Gutenberg. To convert a text file into an RDD, we use the SparkContext.textFile() method. We also apply the recently defined removePunctuation() function using a map() transformation to strip out the punctuation and change all text to lowercase. Since the file is large we use take(15), so that we only print 15 lines.¶

In [66]:
# Just run this code
import os.path
baseDir = os.path.join('data')
inputPath = os.path.join('cs100', 'lab1', 'shakespeare.txt')
fileName = os.path.join(baseDir, inputPath)

shakespeareRDD = (sc
.textFile(fileName, 8)
.map(removePunctuation))
print '\n'.join(shakespeareRDD
.zipWithIndex()  # to (line, lineNum)
.map(lambda (l, num): '{0}: {1}'.format(num, l))  # to 'lineNum: line'
.take(15))

0: 1609
1:
2: the sonnets
3:
4: by william shakespeare
5:
6:
7:
8: 1
9: from fairest creatures we desire increase
10: that thereby beautys rose might never die
11: but as the riper should by time decease
12: his tender heir might bear his memory
13: but thou contracted to thine own bright eyes
14: feedst thy lights flame with selfsubstantial fuel


#### Before we can use the wordcount() function, we have to address two issues with the format of the RDD:¶

• #### The first issue is that that we need to split each line by its spaces.
• #### The second issue is we need to filter out empty lines.

#### Apply a transformation that will split each element of the RDD by its spaces. For each element of the RDD, you should apply Python's string split() function. You might think that a map() transformation is the way to do this, but think about what the result of the split() function will be.¶

In [67]:
# TODO: Replace <FILL IN> with appropriate code
shakespeareWordsRDD = shakespeareRDD.flatMap(lambda x:x.split(' '))
shakespeareWordCount = shakespeareWordsRDD.count()
print shakespeareWordsRDD.top(5)
print shakespeareWordCount

[u'zwaggerd', u'zounds', u'zounds', u'zounds', u'zounds']
927631

In [68]:
# TEST Words from lines (4d)
# This test allows for leading spaces to be removed either before or after
# punctuation is removed.
Test.assertTrue(shakespeareWordCount == 927631 or shakespeareWordCount == 928908,
'incorrect value for shakespeareWordCount')
Test.assertEquals(shakespeareWordsRDD.top(5),
[u'zwaggerd', u'zounds', u'zounds', u'zounds', u'zounds'],
'incorrect value for shakespeareWordsRDD')

1 test passed.
1 test passed.


#### The next step is to filter out the empty elements. Remove all entries where the word is ''.¶

In [69]:
# TODO: Replace <FILL IN> with appropriate code
shakeWordsRDD = shakespeareWordsRDD.filter(lambda x: len(x)!=0)
shakeWordCount = shakeWordsRDD.count()
print shakeWordCount

882996

In [70]:
# TEST Remove empty elements (4e)
Test.assertEquals(shakeWordCount, 882996, 'incorrect value for shakeWordCount')

1 test passed.


#### Use the wordCount() function and takeOrdered() to obtain the fifteen most common words and their counts.¶

In [71]:
# TODO: Replace <FILL IN> with appropriate code
top15WordsAndCounts =wordCount(shakeWordsRDD).takeOrdered(15, lambda (x,v):-v)
print '\n'.join(map(lambda (w, c): '{0}: {1}'.format(w, c), top15WordsAndCounts))

the: 27361
and: 26028
i: 20681
to: 19150
of: 17463
a: 14593
you: 13615
my: 12481
in: 10956
that: 10890
is: 9134
not: 8497
with: 7771
me: 7769
it: 7678

In [73]:
# TEST Count the words (4f)
Test.assertEquals(top15WordsAndCounts,
[(u'the', 27361), (u'and', 26028), (u'i', 20681), (u'to', 19150), (u'of', 17463),
(u'a', 14593), (u'you', 13615), (u'my', 12481), (u'in', 10956), (u'that', 10890),
(u'is', 9134), (u'not', 8497), (u'with', 7771), (u'me', 7769), (u'it', 7678)],
'incorrect value for top15WordsAndCounts')

1 test passed.