<jupyter_start><jupyter_text>Introduction to RDDs - Apache SparkThis notebook provides an introduction to Apache Spark RDD API using PySpark. Press Shift-Enter to execute the code. You can use code completion by using tab.During the exercises you may want to refer to [The PySpark documentation](https://spark.apache.org/docs/latest/api/python/pyspark.html) for more information on possible transformations and actions. We will provide links to the documentation when we introduce methods on RDDs.  The SparkContextThe SparkContext contains all the information about the way Spark is set up. When running on a cluster, the SparkContext contains the address of the cluster and will make sure operations on RDDs will be executed there. In the cell below, we create a [`SparkContext`](https://spark.apache.org/docs/latest/api/python/pyspark.htmlpyspark.SparkContext) using `local mode`. This means that Spark will run locally, not on a cluster. It will offer some form of parallelism by making use of the various cores it has available.Note, that Spark is best used in `cluster mode` where it will run on many machines simultaneously. `Local mode` is only meant for training or testing purposes. However, Spark works quite well in local mode and can be quite powerful. In order to run locally developed code on a cluster, the only thing that needs to be changed is the `SparkContext` and paths to in- and output files.Even when working in `local mode` it is important to think of an RDD as a data structure that is distributed over many machines on a cluster, and is not available locally. The machine that contains the `SparkContext` is called the *driver*. The SparkContext will communicate with the cluster manager to make sure that the operations on RDDs will run on the cluster in the form of *workers*. It is important to realize that the driver is a separate entity from the nodes in the cluster. You can consider the notebook as being the driver.<jupyter_code># Initialize Spark
from pyspark import SparkContext, SparkConf
import nltk
nltk.download('punkt')
nltk.download('averaged_perceptron_tagger')

if not 'sc' in globals(): # This 'trick' makes sure the SparkContext sc is initialized exactly once
    conf = SparkConf().setMaster('local[*]')  # Spark will use all cores (*) available
    sc = SparkContext(conf=conf)<jupyter_output>[nltk_data] Downloading package punkt to /home/jovyan/nltk_data...
[nltk_data]   Unzipping tokenizers/punkt.zip.
[nltk_data] Downloading package averaged_perceptron_tagger to
[nltk_data]     /home/jovyan/nltk_data...
[nltk_data]   Unzipping taggers/averaged_perceptron_tagger.zip.
21/11/10 14:00:54 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Using Spark's default log4j profile: org/apache/spark/log4j-defaults.properties
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).<jupyter_text>Creating an RDDThere are three ways to create an RDD: by transforming an existing one, by reading in data, or by creating an RDD based on a local data structure. We show this last option below.A Python list containing some words is used to create an RDD by calling [`parallelize`](https://spark.apache.org/docs/latest/api/python/pyspark.htmlpyspark.SparkContext.parallelize), a method of `SparkContext`. This list is very small and will not benefit from the parallelism of Spark. We then print the number of records in the RDD, by calling the `count()` method.<jupyter_code>words_list = ['Dog', 'Cat', 'Rabbit', 'Hare', 'Deer', 'Gull', 'Woodpecker', 'Mole']
words_rdd = sc.parallelize(words_list)
print(words_rdd.count())<jupyter_output>[Stage 0:>                                                          (0 + 1) / 1]<jupyter_text>Map transformation There are two kinds of operations on RDDs: transformations and actions. Transformations take as input an RDD and produce as output another RDD (you cannot change an existing RDD, they are immutable). Computation of transformations is deferred until an *action* is executed. An action does not return an RDD, but instead returns data to the driver (for example in the form of a Python list), or writes data to disk or a database.This *laziness* of executing transformations allows Spark to optimize computations. Only when the user wants real output, the framework will start to compute.One of the most used transformations is [`map`](https://spark.apache.org/docs/latest/api/python/pyspark.htmlpyspark.RDD.map). This is very similar to the `Map` in MapReduce. The Spark version of `Map` is a method called `map` defined on an RDD, and takes as input a single function. This function will be applied to each element in the RDD, and Spark will put the result of the application in the output RDD.First, we present a simple Python function that takes a single word as argument and returns the word with an 's' added to it. In the next step we will use this function in a map transformation of the `words_rdd`.Take a look at the function definition below and execute it.<jupyter_code>def make_plural(word):
    return word + 's'

# Let's see if it works

print(make_plural('cat'))<jupyter_output>cats<jupyter_text>Next, we want to use the `make_plural` function as input for the `map` transformation on `words_rdd`.The action [collect()](https://spark.apache.org/docs/latest/api/python/pyspark.html?highlight=collectpyspark.RDD.collect) transfers the content of the RDD to the driver. The result of `collect()` will then be available to our local environment in Python. It is not an RDD but a Python list!Note, that a large RDD may be scattered over many machines. In such a case calling `collect()` may not be a good idea, since it can take quite some time to retrieve all RDD parts.  Exercise 1In the cell below enter the name of the function that map should apply to each element of the RDD in order to end up with an RDD of words in plural form.<jupyter_code>plural_rdd = words_rdd.map(make_plural)

print(plural_rdd.collect())<jupyter_output>['Dogs', 'Cats', 'Rabbits', 'Hares', 'Deers', 'Gulls', 'Woodpeckers', 'Moles']<jupyter_text>Using lambda functionsWe can achieve the same functionality by using lambda functions. In this case we define `make_plural` not using `def` as we did above, but as an anonymous function that we define inside `map` directly. This is the main benefit of using lambda functions: all our processing logic is directly visible in the transformations we're applying.  Exercise 2Provide a lambda function in the cell below, that will pluralize all elements in the RDD.<jupyter_code>lambda_plural_rdd = words_rdd.map(<FILL IN>)

print(lambda_plural_rdd.collect())<jupyter_output><empty_output><jupyter_text>Exercise 3Another transformation is [filter()](https://spark.apache.org/docs/latest/api/python/pyspark.html?highlight=filterpyspark.RDD.filter). It takes as argument a predicate function (a function that is evaluated to true or false), and applies the predicate to all elements of the RDD. Only elements that are evaluated to true by the filter function, will be passed on to the output RDD.Use the [filter()](https://spark.apache.org/docs/latest/api/python/pyspark.html?highlight=filterpyspark.RDD.filter) method of RDD to keep only words with a length larger than three. Use a lambda function to write a predicate that does this. Next, [count()](https://spark.apache.org/docs/latest/api/python/pyspark.html?highlight=countpyspark.RDD.count) the number of words. Like `collect`, [count()](https://spark.apache.org/docs/latest/api/python/pyspark.html?highlight=countpyspark.RDD.count) is an action. Remember that actions trigger Sparks computations. Transformations are evaluated lazily and their computation is deferred until an action is called.There should be 6 words that pass the filter.<jupyter_code>filtered_rdd = words_rdd.filter(<FILL IN>)
filtered_rdd.collect()<jupyter_output><empty_output><jupyter_text>Exercise 4Let's do another `map` transformation on words_rdd. For each word in word_rdd determine its length, again using a lambda function.<jupyter_code>word_lengths = <FILL IN>.collect()
print(word_lengths)<jupyter_output><empty_output><jupyter_text>FlatMap transformationSometimes, the result of a `map` operation is a list of elements rather than a single element. Consider the following example, where we have a list of sentences, and we split each sentence:<jupyter_code>sentences = sc.parallelize([
    'this is a sentence',
    'and this is another one'
])
sentences_rdd = sentences.map(str.split)
sentences_rdd.collect()<jupyter_output><empty_output><jupyter_text>Each element in the RDD returned by `map` is a list of words. Consequently, the result of `collect` is a list of lists, each list containing the sentences' words. Hence, a `count` of this RDD will return two:<jupyter_code>sentences_rdd.count()<jupyter_output><empty_output><jupyter_text>If we want to count the number of words instead, or work directly with the words, we will need to _flatten_ the list of lists into a single list. To do so, we will substitute `flatMap` for `map`. Like `map`, `flatMap` will apply the supplied function to each element in the RDD. In addition to `map`, though, it will _flatten_ the result of the operation such that a list of lists becomes a list:<jupyter_code>sentences.flatMap(str.split).collect()<jupyter_output><empty_output><jupyter_text>Pair RDDsPair RDDs are very important within the Spark RDD API. Each element of a Pair RDD is a pair (or tuple) `(x,y)` where `x` is interpreted as being the key and `y` as the value. Spark offers quite a number of `...byKey` and `...byValues` methods that operate on pair RDDs. As we will see, these methods can be used to define functions per key, very similar to Hadoop's MapReduce.Keys can be of any *hashable* type, which means all primitive types (numbers, strings, etc.), tuples, **but not lists or dictionaries**. Values can be of any type. Below we define a Python string variable called `sonnet`. It is assigned Shakespeare's first sonnet in the form of a single line of text. The character `\` is used to let Python ignore the new line character. Execute the cell, otherwise the variable is not declared and assigned a value.<jupyter_code>sonnet = "From fairest creatures we desire increase, \
That thereby beauty\'s rose might never die, \
But as the riper should by time decease, \
His tender heir might bear his memory: \
But thou contracted to thine own bright eyes, \
Feed'st thy light's flame with self-substantial fuel, \
Making a famine where abundance lies, \
Thy self thy foe, to thy sweet self too cruel: \
Thou that art now the world's fresh ornament, \
And only herald to the gaudy spring, \
Within thine own bud buriest thy content, \
And, tender churl, mak'st waste in niggarding: \
Pity the world, or else this glutton be, \
To eat the world\'s due, by the grave and thee."<jupyter_output><empty_output><jupyter_text>Python magicFrom this text we first remove punctuation. The next cell is just Python. You may want to skip this if your focus is just on Spark, but don't forget to execute the cell.`maketrans()` is a Python method on strings that very efficiently can make character substitutions. Below we use it to remove all punctuation characters. The curly braces indicate a dictionary, and the expression within it, is called a comprehension. The result is a dictionary of key-value pairs, called table, where the key is a punctuation character and the value is `None`. When making substitutions by means of `translate` this table then removes all the entries that have a `None` value.<jupyter_code>import string

# The following line creates a translation table
table = str.maketrans({key: None for key in string.punctuation})

# Do a sample translation
s = "string. With. Punctuation?"
print(s.translate(table))<jupyter_output><empty_output><jupyter_text>Parallelizing the textIn the next cell a lot is happening in one line. The text above is first translated - which in this case means that each punctuation character is removed. Then on the result, the `lower()` method is applied. (This is a Python method on strings.) This puts a string in lowercase letters. Then this result is `split()`, meaning that the text is split in individual words. (Also a Python method on strings). This results in a list of words, all lowercase, with no punctuation. This is input to the `parallelize()` method which turns it into an RDD.*Calling consecutive methods by using dot-notation is called chaining. It is possible of course to execute these steps individually, but chaining can be very convenient, especially in Spark. Consider the individual steps: first parallelize the text, then map the resulting RDD to remove the punctuation, then map the resulting RDD to lowercase the text and then map the resulting RDD of that step to split the data... Doing this instead by chaing methods safes a lot of typing.* To show just the 5 first elements, we use Spark's [`take()`](https://spark.apache.org/docs/latest/api/python/pyspark.htmlpyspark.RDD.take) action. This limits the amount of data that is sent to the driver.<jupyter_code>text_rdd = sc.parallelize(sonnet.translate(table).lower().split())

text_rdd.take(5)<jupyter_output><empty_output><jupyter_text>Exercise 5What would happen if we wouldn't split the text but directly transform it into an RDD? Try this in the next cell (omit `translate` and `lower` as well).Try to predict what will happen. Remember that a string in Python is very similar to a list. (For a list called `mylist` the first element is given by `mylist[0]`. Similarly `mystring[0]` will return the first character of the string `mystring`.)<jupyter_code>another_rdd = sc.<FILL IN>
another_rdd.collect()<jupyter_output><empty_output><jupyter_text>Exercise 6We are going to count the words in `text_rdd`. As a first step, transform every word in `text_rdd` into a tuple `(, 1)`. Use a lambda function.<jupyter_code>pair_rdd = text_rdd.<FILL IN>
pair_rdd.take(5)<jupyter_output><empty_output><jupyter_text>Exercise 7There is an *action* called [countByKey](https://spark.apache.org/docs/latest/api/python/pyspark.htmlpyspark.RDD.countByKey) that performs the counting and returns it as a Python dictionary.Use it below to see the counts.<jupyter_code>word_counts = <FILL IN>

# Below is some Python code that will nicely print the word_counts dictionary

sorted_word_counts = sorted(word_counts.items(), key=lambda x: -x[1])

for word, count in sorted_word_counts[:10]:
    print(word, count)<jupyter_output><empty_output><jupyter_text>reduceByKey`countByKey` is an _action_, returning the word counts as a dictionary instead of an RDD. When using `countByKey` with a large number of counts, the dictionary that is sent back to the driver may not fit in memory.If we want to count words and keep the result into an RDD we have to use the [reduceByKey](https://spark.apache.org/docs/latest/api/python/pyspark.html?highlight=reducebykeypyspark.RDD.reduceByKey) *transformation*.This transformation works almost exactly like Reduce in Hadoop's MapReduce. It expects the RDD to consist of key-value pairs and it will perform a reduce operation *per key*.As input [reduceByKey](https://spark.apache.org/docs/latest/api/python/pyspark.html?highlight=reducebykeypyspark.RDD.reduceByKey) takes a *two-argument function* that will be applied on the values when they are grouped by key.  Exercise 8Create a lambda function that does the counting and forms the input for `reduceByKey`.<jupyter_code># Note that reduceByKey takes in a function that accepts two values and returns a single value
# The function that is input to reduceByKey only works on the values. Spark will execute this function per key

word_counts = pair_rdd.reduceByKey(<FILL IN>)
print(word_counts.collect())<jupyter_output><empty_output><jupyter_text>Instead of using `collect` we can use [takeOrdered](https://spark.apache.org/docs/latest/api/python/pyspark.htmlpyspark.RDD.takeOrdered) to see the most frequent words first.Below we show 10 elements from the RDD. The elements are pairs and we sort them by the second element (denoted by `x[1]` in the lambda function. The minus indicates descending order.<jupyter_code>word_counts.takeOrdered(10, lambda x: -x[1])<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.07341694707861732, 'fraction_numerical': 0.002569593147751606, 'mean_word_length': 5.069810620126253, 'pattern_counts': {'":': 0, '<': 76, '<?xml version=': 0, '>': 77, 'https://': 14, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '18', 'dir': 'jupyter-structured-clean-dedup', 'id': '310100', 'license': 'BSD-3-Clause', 'n_tokens_mistral': 4666, 'n_tokens_neox': 4482, 'n_words': 2145, 'path': '01-spark-rdds.ipynb', 'repo_name': 'lsteffenel/simple-pyspark-binder', 'size': '16345'}