RDD - Transformations - Introduction to PySpark for Data Engineering

RDD Transformations

PySpark RDD Transformations are lazy evaluations used to transform/update RDDs. When executed on an RDD, it results in one or more new RDDs.

Since RDDs are immutable by nature, transformations always create a new RDD without updating an existing one. Therefore, a chain of RDD transformations creates an RDD lineage.

RDD Transformations are Lazy

RDD Transformations are lazy operations, meaning none of the transformations get executed until you call an action on a PySpark RDD. Since RDDs are immutable, any transformations on them result in a new RDD leaving the current one unchanged.

RDD Transformation Types

There are two types of transformations.

Narrow Transformation

Narrow transformations are the result of map() and filter() functions, and these compute data that live on a single partition. This means there will not be any data movement between partitions to execute narrow transformations.

Functions such as map(), mapPartitions(), flatMap(), filter(), union() are some examples of narrow transformations.

Wide Transformation

Wide transformations are the result of groupByKey() and reduceByKey() functions, which compute data that live on many partitions. This means there will be data movements between partitions to execute wide transformations. Since these shuffle the data, they are also called shuffle transformations.

Functions such as groupByKey(), aggregateByKey(), aggregate(), join(), repartition() are some examples of wide transformations.

Note: When compared to narrow transformations, wide transformations are expensive operations due to shuffling.

PySpark RDD Transformation Methods

Following are some of the transformations.

| TRANSFORMATION METHODS | USAGE AND DESCRIPTION | | ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | cache() | Caches the RDD | | filter() | Returns a new RDD after applying the filter function on the source dataset. | | flatMap() | Returns flattened map meaning if you have a dataset with an array, it converts each element in an array as a row. In other words, it returns 0 or more items in output for each element in the dataset. | | map() | Applies transformation function on the dataset and returns the same number of elements in the distributed dataset. | | mapPartitions() | Similar to map, but executes transformation function on each partition. This gives better performance than the map function. | | mapPartitionsWithIndex() | Similar to map Partitions, but also provides func with an integer value representing the index of the partition. | | randomSplit() | Splits the RDD by the weights specified in the argument. For example rdd.randomSplit([0.7, 0.3]) | | union() | Combines elements from the source dataset and the argument and returns the combined dataset. This is similar to the union function in Math set operations. | | sample() | Returns the sample dataset. | | intersection() | Returns the dataset which contains elements in both the source dataset and an argument. | | distinct() | Returns the dataset by eliminating all duplicated elements. | | repartition() | Returns a dataset with a number of partitions specified in the argument. This operation reshuffles the RDD randomly. It could either return lesser or more partitioned RDD based on the input supplied. | | coalesce() | Similar to repartition, but operates better when we want to decrease the partitions. Betterment is achieved by reshuffling the data from fewer nodes compared with all nodes by repartition. |

PySpark RDD Transformations with Examples

In this section, I will explain a few RDD Transformations with examples.

First, create a text file containing the text you want to analyze. For this example, let's assume you have a text file named test.txt with the following content:

Project Gutenberg’s
Alice’s Adventures in Wonderland
Project Gutenberg’s
Adventures in Wonderland
Project Gutenberg’s

# Create SparkSession
from pyspark.sql import SparkSession

spark = SparkSession.builder.master("local[3]").appName("PyAIStudyLabs").getOrCreate()

# Create RDD
sc = spark.sparkContext
rdd = sc.textFile("/path/test.txt")

flatMap() Transformation

flatMap() transformation flattens the RDD after applying the function and returns a new RDD. In the example below, it splits each record by space in an RDD and finally flattens it. The resulting RDD consists of a single word on each record.

# Apply flatMap()
rdd2 = rdd.flatMap(lambda x: x.split(" "))
for element in rdd2.collect():
    print(element)

map() Transformation

map() transformation is used to apply any complex operations like adding a column or updating a column, etc. The output of map transformations would always have the same number of records as the input.

In our word count example, we are adding a new column with the value 1 for each word. The result of the RDD is like PairRDDFunctions which contains key-value pairs, with the word of type String as Key, and 1 of type Int as value. For clarity, I've defined the rdd3 variable with type:

# Apply map()
rdd3 = rdd2.map(lambda x: (x, 1))
for element in rdd3.collect():
    print(element)

filter() Transformation

filter() transformation is used to filter the records in an RDD. In our example, we are filtering all words starting with “a”.

# Apply filter()
rdd4 = rdd3.filter(lambda x: x[0].startswith("a"))
for element in rdd4.collect():
    print(element)

reduceByKey() Transformation

reduceByKey() merges the values for each key with the function specified. In our example, it reduces the word string by applying the sum function on the value. The result of our RDD contains unique words and their count.

# Apply reduceByKey()
rdd5 = rdd3.reduceByKey(lambda a, b: a + b)
for element in rdd5.collect():
    print(element)

sortByKey() Transformation

sortByKey() transformation is used to sort RDD elements on key. In our example, first, we convert RDD[(String,Int)] to RDD[(Int,String)] using map transformation and apply sortByKey, which ideally does sort on an integer value. Finally, we use a foreach loop with a print statement to print all words in RDD and their count as key-value pairs to the console.

# Apply sortByKey()
rdd6 = rdd5.map(lambda x: (x[1], x[0])).sortByKey()

# Print rdd6 result to console
for x in rdd6.collect():
    print(x)

Example Output:

(1, 'Alice’s')
(2, 'in')
(2, 'Adventures')
(2, 'Wonderland')
(3, 'Gutenberg’s')
(3, 'Project')

RDD - Transformations