Get started with PySpark and Resilient Distributed Datasets (RDDs)

RDDs with PySpark

The `reduce()` function we used previously is a powerful aggregation tool, but there are limitations to the operations it can apply to RDDs. Namely, `reduce()` must be **commutative** and **associative** due to the nature of parallelized computation. 

You’ve probably heard of both those terms back in elementary math class, and they probably make sense to you in that context. However, what do they mean in Spark?

Well, it all ties back to the fact that Spark operates in parallel &mdash; tasks that have commutative and associative properties allow for parallelization. 
* The commutative property allows for all parallel tasks to execute and conclude without waiting for another task to complete. 
* The associative property allows Spark to partition and distribute our data to multiple nodes because the result will stay the same no matter how tasks are grouped. 

Let’s try to break that down a bit further with math!
No matter how you switch up or break down summations, they’ll always have the same result thanks to the commutative and associative properties:
```tex
1+2+3+4+5 = (3+4+5)+(1+2) = (4)+(2+5+1)+(3) = 15
```
However, this is not the case with division:
```tex
1\div2\div3\div4\div5 \neq (3\div4\div5)\div(1\div2) \neq 4\div(2\div5\div1)\div3
```
The flowchart represents one of the possible ways that our list was partitioned into three nodes and ultimately summed. No matter how our data was partitioned or which summations were completed first, the answer will be 15.

![title](https://static-assets.codecademy.com/Courses/big-data-pyspark/rdd-add.svg)

This shows that the commutative and associative properties enable parallel processing because it gives us two very important concepts: the output doesn’t depend on the order in which tasks complete (commutative) nor does it depend on how the data is grouped (associative).

Associative and Commutative Properties

Diagram of how data gets partitioned among workers in spark. Shows how it starts with an RDD, then each partition creates an original and a copy. That is then distributed to a cluster of nodes. 

Apache Spark is a framework that allows us to work with big data. But how do we tell Spark what to do with our data? In this lesson, we'll get familiar with using PySpark (the Python API for Spark) to load and transform our data in the form of **RDDs** &mdash; **resilient distributed datasets**. 

RDDs are the foundational data structures of Spark. Newer Spark structures like **DataFrames** are built on top of RDDs. While DataFrames are more commonly used in industry, RDDs are not deprecated and are still called for in certain circumstances. For example, RDDs are useful for processing unstructured data, such as text or images, that don't fit nicely in the tabular structure of a DataFrame.

So what exactly is an RDD? According to our friends at Apache, the formal definition of an RDD is “a fault-tolerant collection of elements partitioned across the nodes of the cluster that can be operated on in parallel.” Those are some complicated words! Let’s break down the three key properties of RDDs together:

- **Fault-tolerant** or **resilient**: data is copied and recoverable in the event of failure
- **Partitioned** or **distributed**: datasets are split up across the nodes in a cluster
- **Operated on in parallel**: tasks are executed on all the chunks of data at the same time

Now that we have a bit more context as to what RDDs are, let’s learn how to create one with PySpark in the next exercise!

Resilient Distributed Datasets (RDDs)

Imagine having an RDD containing two-letter state abbreviations. 

```py
# list of states
states = ['FL', 'NY', 'TX', 'CA', 'NY', 'NY', 'FL', 'TX']
# convert to RDD
states_rdd = spark.sparkContext.parallelize(states)
```

However, we want the region instead of the state. Regions are groupings of states based on their geographic location, such as "East" or "South". Currently, our RDD is partitioned in the Spark cluster, and we don’t know which nodes contain data on which states. 

In this situation, we need to send the conversion information to all nodes because it’s very likely that each node will contain multiple distinct states. We can provide each node with information on which states belong in each region. This information that is made available to all nodes is what Spark calls **broadcast variables**. Let’s see how we can implement them to convert the abbreviations! 

Let’s start off by creating a conversion dictionary called `region` that matches each state to its region:

```py
# dictionary of regions
region = {"NY":"East", "CA":"West", "TX":"South", "FL":"South"}
```

We can then broadcast our `region` dictionary and apply the conversion to each element in the RDD with our map function:

```py
# broadcast region dictionary to nodes
broadcast_var = spark.sparkContext.broadcast(region)
# map regions to states
result = states_rdd.map(lambda x: broadcast_var.value[x])
# view first four results
result.take(4)
# output : [‘South’, ‘East’, ‘South’, ‘West’]
```

This is Spark’s efficient method of sharing variables amongst its nodes (also known as **shared variables**). They ultimately improve performance by decreasing the amount of data transfer overhead because each node already has a cached copy of the required object. However, it should be noted that we would never want to broadcast large amounts of data because the size would be too much to serialize and send through the network.

Broadcast Variables

We’ve broadcasted a dictionary over to your nodes, and everything went well! We’re now curious as to how many "East" versus "West" entries there are. We could attempt to create a couple of variables to keep track of the counts, but we might run into serialization and overhead issues when datasets get really big. Thankfully, Spark has another type of shared variable that solves this issue: **accumulator variables**.

Accumulator variables can be updated and are primarily used as counters or sums. Conceptually, they’re similar to the sum and count functions in NumPy.

Let’s see how we can implement accumulator variables by counting the number of distinct regions. Since this will be a new dataset, let’s create an RDD first:
```py
region = ['East', 'East', 'West', 'South', 'West', 'East', 'East', 'West', 'North']
rdd = spark.sparkContext.parallelize(region)
```

We’ll start off by initializing the accumulator variables at zero:
```py
east = spark.sparkContext.accumulator(0)
west = spark.sparkContext.accumulator(0)
```

Let’s create a function to increment each accumulator by one whenever Spark encounters 'East' or 'West':
```py
def countCoasts(r):
    if 'East' in r: east.add(1)
    elif 'West' in r: west.add(1)
```

We’ll take the function we created and run it against each element in the RDD.
```py 
rdd.foreach(lambda x: countCoasts(x))
print(east) # output: 4
print(west) # output: 3
```
This seems like a simple concept, but accumulator variables can be very powerful in the right situation. They can keep track of the inputs and outputs of each Spark task by aggregating the size of each subsequent transformation. Instead of counting the number of east or west coast states, we could count the number of NULL values or the resulting size of each transformation. This is important to monitor for data loss. 

This doesn’t mean you should add accumulator variables to everything though. It’s best to avoid using accumulators in _transformations_. Whenever Spark runs into an exception, it will re-execute the tasks. This will incorrectly increment the accumulator. However, Spark will guarantee that this does not happen to accumulators in _actions_.

Accumulators can be great as debugging or summary tools, but they’re not infallible when used in transformations.

Accumulator Variables

The entry point to Spark is called a **SparkSession**. There are many possible configurations for a SparkSession, but for now, we will simply start a new session and save it as `spark`:

```py
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate() 
```

We can use Spark with data stored on a distributed file system or just on our local machine. Without additional configurations, Spark defaults to local with the number of partitions set to the number of CPU cores on our local machine (often, this is four). 

The `sparkContext` within a SparkSession is the connection to the cluster and gives us the ability to create and transform RDDs. We can create an RDD from data saved locally using the `parallelize()` function. We can add an argument to specify the number of partitions, which is generally recommended as 2-4 partitions per machine. Otherwise, Spark defaults to the total number of CPU cores. 
```py
# default setting
rdd_par = spark.sparkContext.parallelize(dataset_name)
```

If we are working with an external dataset, or possibly a large dataset stored on a distributed file system, we can use `textFile()` to create an RDD. Spark's default is to partition the text file in 128 MB blocks, but we can also add an argument to set the number of partitions within the function.
```py
# with partition argument of 10
rdd_txt = spark.sparkContext.textFile("file_name.txt", 10)
```

We can verify the number of partitions in `rdd_txt` using the following line:
```py
rdd_txt.getNumPartitions()
# output: 10
```

Finally, we need to know how to end our SparkSession when we are finished with our work:
```py
spark.stop()
```

Now that we know how to get started with PySpark, let’s introduce the dataset we’ll be working with throughout this lesson and set it up as an RDD!

---

_How to Use Your Jupyter Notebook:_
* _You can run a cell in the Notebook to the right by placing your cursor in the cell and clicking the `Run` button or the `Shift`+`Enter/Return` keys._
* _When you are ready to evaluate the code in your Notebook, press the `Save` button at the top of the Notebook or `command`+`s` keys before clicking the `Test Work` button at the bottom. **Be sure to save your solution code in the cell marked `## YOUR SOLUTION HERE ##` or it will not be evaluated.**_
* _When you are ready to move on, click `Next`._

![screenshot of the buttons at the top of the Jupyter Notebook interface with Save and Run highlighted](https://static-assets.codecademy.com/Courses/big-data-pyspark/Jupyter-buttons.png)

Start Coding with PySpark

Congratulations! You've just finished your first coding adventure with PySpark! In this lesson, we learned that:
* RDDs are the foundational data structure of Spark
* RDDs are fault-tolerant, partitioned, and operated on in parallel
* Transformations are lazy and do not execute until an action is called

We also learned how to:
* Transform and summarize RDDs with transformations and actions
* Send information to all nodes with broadcast variables
* Debug work with accumulator variables

Review

You may have noticed that transformations execute rather quickly! That’s because they didn’t execute at all. Spark executes transformations only when an **action** is called to return a value. This delay is why we call Spark transformations **lazy**. We call the transformations we do in pandas **eager** because they execute immediately.

So, why are Spark transformations lazy? Spark will queue up the transformations to optimize and reduce overhead once an action is called. Let’s say that we wanted to apply a map and filter to our RDD:

```py
rdd = spark.SparkContent.parallelize([1,2,3,4,5])
rdd.map(lambda x: x+1).filter(lambda x: x>3)
```

Instead of following the order that we called the transformations, Spark might load the values greater than 3 into memory first and perform the map function last. This swap will save memory and time because Spark loaded fewer data points and mapped the lambda to fewer elements.

In the last exercise, Spark executed our transformations only when the action `collect()` was called to return the entire contents of the new RDD as a list. We generally don't want to use `collect()` to pull large amounts of data into memory, so we can use `take(n)` to view the first **_n_** elements of a large RDD.

```py
# input RDD [1,2,3,4,5]
rdd.take(3)
```
```
[1, 2, 3]
```

We can use the action `reduce()` to return fewer elements of our RDD by applying certain operators. For example, say we want to add up all the values in the RDD. We can use `reduce()` with a lambda to add each element sequentially.

```py
# input RDD [1,2,3,4,5]
rdd.reduce(lambda x,y: x+y)
```
```
15
```

`reduce()` is powerful because it allows us to apply many arbitrary operations to an RDD &mdash; it unbinds us from searching for library functions that might not exist. However, it certainly has limitations, which we’ll dive into in the next exercise.

The key thing about actions is that, like transformations, they take an RDD as input, but they will always output a value instead of a new RDD.

Actions

Many of the Spark functions we use on RDDs are similar to those we regularly use in Python. We can also use lambda expressions within RDD functions. Lambdas allow us to apply a simple operation to an object in a single line without defining it as a function. Check out the following example of a lambda expression that adds the number 1 to its input.

```py
add_one = lambda x: x+1 # apply x+1 to x
print(add_one(10)) # this will output 11
```

Let's introduce a couple of PySpark functions that we may already be familiar with:

`map()` applies an operation to each element of the RDD, so it's often constructed with a lambda expression. This map example adds 1 to each element in our RDD:

```py
rdd = spark.SparkContent.parallelize([1,2,3,4,5])
rdd.map(lambda x: x+1)
# output RDD [2,3,4,5,6]
```

If our RDD contains tuples, we can map the lambda expression to the elements with a specific index value. The following code maps the lambda expression to just the first element of each tuple but keeps the others in the output:

```py
# input RDD [(1,2,3),(4,5,6),(7,8,9)]
rdd.map(lambda x: (x[0]+1, x[1], x[2]))
# output RDD [(2,2,3),(5,5,6),(8,8,9)]
```

`filter()` allows us to remove or keep data conditionally. If we want to remove all `NULL` values in the following RDD, we can use a lambda expression in our filter:

```py
# input RDD [1,2,NULL,4,5]
rdd.filter(lambda x: x is not None)
# output RDD [1,2,4,5]
```

You may have noticed that each function took an RDD as input and returned an RDD as output. In Spark, functions with this behavior are called **transformations**. You can find more transformations in [the official Spark documentation](https://spark.apache.org/docs/latest/rdd-programming-guide.html#transformations).

We have one final note about transformations: we can only view the contents of an RDD by using a special function like `collect()`, which will print the data stored in the RDD. So to view the new RDD in the previous example, we would run the following:

```py
rdd.filter(lambda x: x is not None).collect()
```
```
[1,2,4,5]
```

Let's try working with some transformations!