A Comprehensive Guide to PySpark RDD Operations

Resilient Distributed Dataset or RDD in a PySpark is a core data structure of PySpark. PySpark RDD’s is a low-level object and are highly efficient in performing distributed tasks. This article will not involve the basics of PySpark such as the creation of PySpark RDDs and PySpark DataFrames. If you are not aware of these terms, I would highly recommend reading my previous article on PySpark here.

PySpark RDD has a set of operations to accomplish any task. These operations are of two types:

1. Transformations

2. Actions

Transformations are a kind of operation that takes an RDD as input and produces another RDD as output. Once a transformation is applied to an RDD, it returns a new RDD, the original RDD remains the same and thus are immutable. After applying the transformation, it creates a Directed Acyclic Graph or DAG for computations and ends after applying any actions on it. This is the reason they are called lazy evaluation processes.

Actions are a kind of operation which are applied on an RDD to produce a single value. These methods are applied on a resultant RDD and produces a non-RDD value, thus removing the laziness of the transformation of RDD.

To conclude in Layman’s Terms, Transformations are applied on an RDD to give another RDD. While Actions are performed on an RDD to give a non-RDD value.

F purposes, we will perform all the following operations in Google Colab. To perform the PySpark RDD Operations, we need to perform some prerequisites in our local machine. If you are also practicing in your local machine, you can follow the following prerequisites.

!pip install pyspark

Next, we will initialize a SparkContext to perform the operations:

from pyspark import SparkContext
sc = SparkContext.getOrCreate()

If you are unable to complete the prerequisites due to any of certain reasons, please refer to the PySpark Installation Guide for Local machines.

Now, we have SparkContext ready with us, we get to perform all the Actions and Transformations coming next.

Actions in PySpark RDDs

5. The .reduce() Action

The .reduce() Actiontakes two elements from the given RDD and operates. This operation is performed using an anonymous function or lambda. For example, if we want to add all the elements from the given RDD, we can use the .reduce() action.

reduce_rdd = sc.parallelize([1,3,4,6])
print(reduce_rdd.reduce(lambda x, y : x + y))

On executing this code, we get:

Here, we created an RDD, reduce_rdd using .parallelize() method of SparkContext. We used the .reduce action on reduce_rdd with an enclosed anonymous function or lambda. Here, the lambda adds all the elements of the given RDD and prints the sum.

 

6. The .saveAsTextFile() Action

The .saveAsTextFile() Action is used to serve the resultant RDD as a text file. We can also specify the path to which file needed to be saved. This helps in saving our results especially when we are working with a large amount of data.

save_rdd = sc.parallelize([1,2,3,4,5,6])
save_rdd.saveAsTextFile('file.txt')

On executing this code, we get:

Here, we created an RDD, save_rdd using the .parallelize() method of SparkContext. We used the .saveAsTextFile() action on save_rdd to save it into our directory with the name passed as an argument in it as a string type. The .saveAsTextFile() generates a directory with the given argument. Inside the directory, several parts of the file will be created based on the size of the file.

Transformations in PySpark RDDs

Transformations are the kind of operations that are performed on an RDD and return a new RDD. Few of these methods work almost similarly to the functions already present in Python. To learn more about Transformations, refer to the Spark Documentation here.

Now, Let’s look at some of the essential Transformations in PySpark RDD:

1. The .map() Transformation

As the name suggests, the .map() transformation maps a value to the elements of an RDD. The .map() transformation takes in an anonymous function and applies this function to each of the elements in the RDD. For example, If we want to add 10 to each of the elements present in RDD, the .map() transformation would come in handy. This operation saves time and goes with the DRY policy. Let’s understand this with an example:

my_rdd = sc.parallelize([1,2,3,4])
print(my_rdd.map(lambda x: x+ 10).collect())

On executing this, we get:

Here, we created an RDD, my_rdd using the .parallelize() method of SparkContext. Since the .map() is a transformation, it returns a new RDD, thus we used the .collect() action to extract all the resultant elements in a Python list. Here, the anonymous function or lambda performs the same as it works in Python.

2. The .filter() Transformation

A .filter() transformation is an operation in PySpark for filtering elements from a PySpark RDD. The .filter() transformation takes in an anonymous function with a condition. Again, since it’s a transformation, it returns an RDD having elements that had passed the given condition. For example, we want to return only an even number of elements, we can use the .filter() transformation.

Let’s understand this with an example:

filter_rdd = sc.parallelize([2, 3, 4, 5, 6, 7])
print(filter_rdd.filter(lambda x: x%2 == 0).collect())

On executing this code, we get:

Here, we first created an RDD, filter_rdd using the .parallelize() method of SparkContext. Then we used the anonymous function lambda to filter the even numbers from our RDD filter_rdd. Since .filter() transformation returns a new RDD, we used the .collect() action to extract all the resultant elements in a list.

We can also filter strings from a certain text present in an RDD. For example, If we want to check the names of persons from a list of guests starting with a certain alphabet, we can use the .filter() for this operation as well. Let’s understand this with an example:

filter_rdd_2 = sc.parallelize(['Rahul', 'Swati', 'Rohan', 'Shreya', 'Priya'])
print(filter_rdd_2.filter(lambda x: x.startswith('R')).collect())

On executing this code, we get:

Here, we first created an RDD, filter_rdd_2 using the .parallelize() method of SparkContext. Then we used the .filter() transformation on it to filter the elements of our RDD that start with ‘R‘. We used the .collect() action on the resultant RDD to get all the desired elements in a list.

3. The .union() Transformation

The .union() transformation combines two RDDs and returns the union of the input two RDDs. This can be helpful to extract elements from similar characteristics from two RDDs into a single RDD.

Let’s understand this with an example:

union_inp = sc.parallelize([2,4,5,6,7,8,9])
union_rdd_1 = union_inp.filter(lambda x: x % 2 == 0)
union_rdd_2 = union_inp.filter(lambda x: x % 3 == 0)
print(union_rdd_1.union(union_rdd_2).collect())

On executing this code, we get:

Here, we first created an RDD, union_inp using the .parallelize() method of SparkContext. Then we created two more RDDs union_rdd_1 and union_rdd_2 using the .filter() method on input RDD. At last, we created the union of the two filtered RDDs using the .union() transformation.

4. The .flatMap() Transformation

The .flatMap() transformation peforms same as the .map() transformation except the fact that .flatMap() transformation return seperate values for each element from original RDD.

Let’s understand this with an example:

flatmap_rdd = sc.parallelize(["Hey there", "This is PySpark RDD Transformations"])
(flatmap_rdd.flatMap(lambda x: x.split(" ")).collect())

On executing this, we get:

Here, we first created an RDD, flatmap_rdd using the .parallelize() method and added two strings to it. Then, we applied the .flatMap() transformation to it to split all the strings into single words. This worked the same as the .split() method in Python lists. Then we used the .collect() method to extract all the resultant elements in a list.

PySpark Pair RDD Operations

Conclusion

In this guide, we learned about PySpark RDDs and it’s operations which are widely used. In general, we looked at two types of operations, Transformation, and Actions and the different methods involved in it. For each of RDD and Pair RDD, we looked at a different set of Actions and Transformations. There is a certain number of Transformations that needed to apply only on Pair RDD. Actions discussed for RDDs are versatile and can be used on Pair RDD as well. We also discussed a single action for Pair RDD which is again, exclusive to only Pair RDD and cannot be used for normal RDD as it requires data in key-value [pair type.

Each type of Transformation or Action plays an important role in itself and one can apply them based on the tasks these operations can accomplish. Few more actions can be used based on the requirement of the project. All the discussed operations are very popular and are used in almost every task of Big Data Analysis.

Read my previous articles on PySpark :

1. Beginner’s Guide To Create PySpark DataFrame

2. Essential PySpark DataFrame Column Operations that Data Engineers Should Know

About the Author

Connect with me on LinkedIn.

For any suggestions or article requests, you can email me here.

Check out my other Articles Here and on Medium

You can provide your valuable feedback to me on LinkedIn.

Thanks for giving your time!

Source: https://www.analyticsvidhya.com/blog/2021/10/a-comprehensive-guide-to-pyspark-rdd-operations/