RDD - Transformations - Apache Spark Tutorial with Scala Examples

RDD Transformations

Spark RDD Transformations are lazy evaluation and is used to transform/update RDD. When executed on RDD, it results in a single or multiple new RDD.

Since RDD are immutable in nature, transformations always create a new RDD without updating an existing one hence, 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 Spark 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 are transformations.

Narrow Transformation

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

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

Wide Transformation

Wide transformations are the result of groupByKey() and reduceByKey() functions and these compute data that live on many partitions meaning 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.

Spark 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() | Comines 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() | Return 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 by 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. |

Spark 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
val spark:SparkSession = SparkSession.builder().master("local[3]")
      .appName("AIStudyLabs.com").getOrCreate()

// Create RDD
val sc = spark.sparkContext 
val rdd:RDD[String] = sc.textFile("/path/test.txt")

// Create RDD
val sc = spark.sparkContext 
val rdd:RDD[String] = sc.textFile("/path/test.txt")

<img height="324" width="527" src="https://yrltjyzsmzdylqvoonwf.supabase.co/storage/v1/object/public/lesson-images/100/181/rdd-transformation-work-count.png" />

flatMap() Transformation

flatMap() transformation flattens the RDD after applying the function and returns a new RDD. In the below example, first, 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() 
val rdd2 = rdd.flatMap(f=>f.split(" "))

map() Transformation

map() transformation is used the apply any complex operations like adding a column, or updating a column e.t.c, 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 a value 1 for each word, the result of the RDD is PairRDDFunctions which contains key-value pairs, word of type String as Key, and 1 of type Int as value. For your understanding, I’ve defined the rdd3 variable with type.

// Apply map()
val rdd3:RDD[(String,Int)] = rdd2.map(m=>(m,1))

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()
val rdd4 = rdd3.filter(a=> a._1.startsWith("a"))

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 value. The result of our RDD contains unique words and their count.

// Apply reduceByKey()
val rdd5 = rdd3.reduceByKey(_ + _)

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. And finally, foreach with println statement prints all words in RDD and their count as key-value pair to the console.

// Apply sortByKey()
val rdd6 = rdd5.map(a=>(a._2,a._1)).sortByKey()

//Print rdd6 result to console
rdd6.foreach(println)

RDD - Transformations