Introduction to RDD
What is RDD?
• RDD stands for Resilient Distributed Datasets
• RDD is the primary data abstraction in Apache Spark and core of
spark.
• As the name suggests RDD is a resilient ( fault – tolerance ) records of
the data that resides on multiple nodes.
Advantage of RDD
• With RDD the creators of spark managed to hide data partitioning and
distribution.
• Learning about RDD by its name i.e.
1. Resilient i.e. Fault Tolerant
2. Distributed i.e. Distributed across multiple node
3. Dataset i.e. Collection of partitioned data
Features of RDD
• In-Memory i.e. Data inside RDD will be kept in memory as long as
possible
• Immutable i.e. data cannot be changed means once we create a RDD
that will be parent RDD so we can have transformation of RDD but
original RDD cannot be changed.
• Lazy Evaluated i.e. delaying the computation until it is required.
• Parallel i.e. data will be processed in parallel.
RDD operations
• RDD supports two type of operations
1. Transformation
2. Actions
• Transformation is changing the state of the data from one state to
another without getting any outputs.
• All the transformations are lazily evaluated i.e. they wait for any
action to be called so actions act as a triggers. When you call an
action you get output.
Ways to create RDD in Spark
• RDD’s are fundamental data structure of Apache spark. It is the basic
abstraction that is being used in Apache Spark.
• There are 4 ways to create RDD:
1. Using [Link] method
2. Using [Link] method
3. To create a RDD from existing RDD using flatMap
4. Create a RDD from existing data frame
1. Using [Link]
method
• To get the collection from the driver program of Scala or python and
the apply [Link] method to get the RDD created.
• This method is only useful for testing purpose we cannot use this in
real-time because entire data is stored in single node which is not
recommended in real-time.
Example
val r1 = [Link](Seq((“A”,1),(“B”,2),(“C”,3)))
• This will create a RDD containing string and int values and type got
inferred because of type inference property of Scala.
• To print the content on console:
[Link](println)
Notation in spark O/P
• Spark uses multiple tasks to process RDDs in parallel.
• The notation (0 + 4) / 4 is a progress indicator:
0: Number of tasks already completed.
4: Number of tasks running or pending.
4: Total number of tasks assigned for execution.
2. Using [Link]
method
• We will read an external file to create a RDD in this method.
• For this method we need to have a file with some content on our
system and path of that file will be provided to create a RDD using
existing file.
Example
val
r2=[Link]("C:/Users/Dell/Desktop/[Link]")
[Link]()
• RDD will be created as per the type of data stored in existing file.
• Collect is an action here which will print the content of file.
3. Create a RDD from existing RDD
using flatMap
• Here we will be using r2 RDD created in last example by applying
flatMap transformation and it will split down the content of r2 RDD.
• Example:
val r3 = [Link](_.split(“ “))
[Link]()
• flatMap is a transformation which is used to read the data word by
word.
4. Create a RDD from existing data
frame
• Here we will be using range instead of data frame to create a RDD.
• Example:
val r4 = [Link](20).toDF().rdd
[Link]()
• toDf() is used to create a data frame out of the range of 20 and .rdd is
used to transform that data frame as an rdd.
Partitions of RDD
• We can create certain partitions of RDD as well.
val rdd=[Link](1 to 20,4)
• Here 4 represents the no. of partitions of an RDD.
• To check the created partitions:
print([Link])
// Create an RDD of numbers 1 to 20, with 4 partitions
• val rdd = [Link](1 to 20, 4)
// Print the number of partitions
println("Number of partitions: " + [Link])
// Collect and display the elements in each partition
println("Elements in each partition:")
[Link]().collect().[Link] {
case (partitionData, partitionIndex) =>
println(s"Partition $partitionIndex: " + [Link](", "))
}
• To check the specific partitions of RDD we can use glom()
[Link]().collect().foreach(println)
• This will return us all the 4 partitions in the form of Hashcode (default
string representation of JAVA array) in the RDD. To visualize the
elements in string form we need to convert that explicitly:
[Link]().collect().foreach(arr=>println([Link](",")))
• Here mkString creates a string from the elements of array collection.
Question
• Write Scala code to parallelize a simple collection (e.g.,
an array or list) into an RDD in Spark.
Solution
# using array
val data = Array(1, 2, 3, 4, 5)
val rdd = [Link](data)
[Link](println)
# using list
val dataList = List("apple", "banana", "orange", "grape", "pineapple")
val rdd = [Link](dataList)
[Link](println)
Question
• Provide Scala code to load a text file named "[Link]" into an RDD
using the textFile() method in Spark.
RDD Transformation
• It is a function that produces new RDD from an existing RDD.
• There are multiple type of transformation methods which can be used
to produce a new RDD.
• However we cannot change the original RDD hence applying the
transformation will create RDD lineage.
RDD Transformation Methods
• Various methods are there for RDD transformation:
1. Map()
2. FlatMap()
3. Filter()
4. Union()
5. Intersection()
6. Distinct()
7. groupByKey()
8. ReduceByKey()
9. sortByKey()
10. Join()
11. Coalesce()
1. map()
• Map function iterates over every line of RDD and splits into new RDD.
• Function will be applied to every element of RDD.
Example
val flat1 = [Link]("C:/Users/Dell/Desktop/[Link]")
val flat2= [Link](r=>[Link](" ")) // split the elements
[Link]() // print
2. flatMap()
• It is similar to map function but the difference is that map return only
one element but flatMap returns list of elements.
• One common example is creating a RDD from existing RDD by splitting
the content into each word.
Example
val flat1 = [Link]("C:/Users/Dell/Desktop/[Link]")
val flat3= [Link](r=>[Link](" "))
[Link]()
3. filter()
• Filter() in spark is quite similar to select operator of DBMS i.e. what
kind of filtrations we need to perform.
• Filter can be used to filter out any specific element from an rdd for
example filtering even number from any rdd.
Example:
val x=[Link]( 1 to 10) // create a rdd with 1 to 10 element
[Link]() // print the elements of rdd
val y =[Link](z=>z%2==0) // filter even numbers
[Link]()
4. Union()
• Union() will combine two elements together.
• So if we are having two rdd’s it can combine both these rdd’s
together.
Example
• Lets first create two rdds
val r1=[Link](Array(1,2,3,4)) // 1st rdd
val r2=[Link](Array(4,6,7,8)) // 2nd rdd
// union
[Link](println) // combined result
5. Intersection()
• Displays the common part between two rdd’s.
• Example:
val r1=[Link](Array(1,2,3,4))
val r2=[Link](Array(4,6,7,8))
val i=[Link](r2) // intersection
[Link](println) // print common result
6. Distinct()
• This function will print the distinct elements only from any rdd.
• Duplicate elements will not be printed.
Example
val dup=[Link](Array(1,2,2,2,2,3,3,3,4,5,6))
val dis=[Link]()
[Link](println) // print distinct elements
7. groupByKey()
• This will group the elements of two rdd’s based on the key and value.
• Example:
val data=[Link](Array((‘a’,1),(‘b’,2),(‘c’,3), (‘a’,5),(‘b’,2)),3)
• 3 denotes how many partition we need to have in background.
val group=[Link]()
[Link](println)
8. reduceByKey()
• Elements will be reduce with the help of the key in rdd.
• For example if we have duplicate elements in any rdd then it can give
us that how many times an element is repeating etc.
Example
val words=Array(“one”,”two”,”two”,”three”)
val red=[Link](words).map(w=>(w,1)).reduceByKey(_+_)
[Link](println)
9 sortByKey()
• Put the elements in sorted order with respect to the keys.
• Example:
val r=[Link](Seq((“A”,2),(“Y”,4),(“Q”,7),(“F”,6)))
val sorted=[Link]()
[Link]().foreach(println)
• Elements will be sorted alphabetically.
10 Join()
• Joins in SQL are used to join two or more tables.
• Similarly here also we will use joins to join two rdd’s.
• Example:
• Lets consider two rdd’s first:
val q= [Link](Array((‘a’,1),(‘b’,2),(‘c’,3)))
val p= [Link](Array((‘a’,4),(‘b’,3),(‘c’,3)))
Join()
• Now lets perform join on two rdd’s i.e. p and q.
val v=[Link](rdd2)
[Link](println)
11. Coalesce()
• Coalesce() is used to reduce number of partitions, it avoid full
shuffling of data. So it will reduce the number of partitions.
• Example:
val rdd = [Link](1 to 100, 6)
println("Partitions before: " + [Link])
val coalescedRdd = [Link](3)
println("Partitions after: " + [Link])
No difference in output but number of partitions will be reduce to 3.
RDD Operations - Actions
• Transformations create RDDs from other RDDs, but when we want to
work with the actual dataset, at that point action is performed. When
the action is triggered after the result, new RDD is not formed like
transformation.
• Thus actions are the spark RDD operations that gives non-RDD values.
List of Actions in Spark RDD
• Following are the actions that can be performed on the spark RDDs.
1. COUNT
2. COLLECT
3. TAKE
4. TOP
5. COUNTBYVALUE
6. REDUCE
List of actions
• val rdd = [Link]("C:/Users/91991/OneDrive/Desktop/[Link]")
• val totalCount = [Link]()
• println(s"Total Count: $totalCount")
• val collectedData = [Link]()
• [Link](println)
Continued….
• val firstFour = [Link](4)
• [Link](println)
• val topThree = [Link](3)
• [Link](println)
• val counts = [Link]()
• [Link]{ case (word, count) => println(s"$word -> $count") }
• val concatenated = [Link]((a, b) => a + " " + b)
• println(concatenated)
• In order to work with actions we need to create a file with some
elements like 1 2 3 4 5 and we need to read that file.
val input=[Link](“C:/Users/Dell/Desktop/[Link]”)
• All the next actions will be performed on the above rdd.
1. Count
• In SQL we use count in order to calculate the number of rows in a
table.
• It will count the number of values available in the rdd.
• Example:
val v1=[Link](_.split(“ “)).count()
• This will return us the number of elements in rdd.
2. Collect
• To see the content of rdd we use collect.
• Example:
val v1=[Link](_.split(“ “)).collect()
• It will return the content of RDD.
3. Take
• Take is similar to the limit operator of SQL.
• In order to limit the output we use Take action.
• Example:
val v1=[Link](_.split(“ “)).take(4)
• This will return first 4 numbers of the RDD.
4. TOP
• Top is somehow similar to orderBy clause which is used to print the
data in ascending or descending order.
• Example:
val v1=[Link](_.split(“ “)).top(3)
• This will return the highest 3 elements as per value.
5. COUNTBYVALUE
• Count gives the number of values in rdd but COUNTBYVALUE will give
the occurrence of each character or value in the file.
• Example:
val v1=[Link](_.split(“ “)).countByValue()
6. REDUCE
• Reduce is very common action which aggregates similar to aggregate
operators of SQL such as max, min etc.
• Example:
val input1=[Link](Array(1,2,3,4))
val output=[Link](_+_)
• This will return the sum of all the array elements.
7. FOREACH
• It do not return any value.
• It executes input function on each element of an RDD.
• Example:
[Link](println)
• It will print the result line by line.
8. takeSample()
• Val r1=[Link](Array(1,2,3,4,5))
• To take a sample from RDD elements
[Link](false,2)
res6: Array[Int] = Array(2, 5)
9. max(), min(), sum()
Val r1=[Link](Array(1,2,3,4,5))
[Link]() // return maximum element
[Link]() // return minimum element
[Link]() // return sum of elements
10. isEmpty()
• Checks if a given rdd is empty or not.
• Returns a Boolean value.
Val r1=[Link](Array(1,2,3,4,5))
[Link]()
11. mean(), stdev()
• Calculates mean and standard deviation of the rdd elements.
Val r1=[Link](Array(1,2,3,4,5))
[Link]()
[Link]()
12. sortBy()
To sort the elements of RDD in asc or desc order
val rdd = [Link](Array(5, 1, 3, 8, 2, 7))
val sortedRddAsc = [Link](x => x, ascending = true)
[Link]().foreach(println)
For Desc
val sortedRddDesc = [Link](x => x, ascending = false)
[Link]().foreach(println)
Practice Question 1
• A retail company stores daily sales transactions in an RDD where each
record is (StoreID, ItemID, Quantity, Price). Retrieve all transactions
from the RDD to inspect them locally.
Solution
val salesRDD = [Link](Seq((1,101,2,20.0), (2,102,1,15.0),
(3,103,3,25.0)))
[Link]()
Practice Question 2
• You are analyzing customer feedback stored in an RDD, where each
record is a comment. Retrieve the first comment for a quick preview.
Solution
• val feedbackRDD = [Link](Seq("Great product!", "Could be
better", "Fast delivery"))
• [Link]()
Practice Question 3
• An e-commerce store has an RDD of purchased product IDs ( e.g.
A,B,A,C,A,D etc.) . Find how many times each product was purchased.
Solution
• val purchasesRDD = [Link](Seq("A", "B", "A", "C", "A", "B", "D",
"C", "B", "B"))
• [Link]()
Practice Question 4
• A student dataset contains exam scores in an RDD. Retrieve the top 3
scores.
Solution
• val scoresRDD = [Link](Seq(85, 92, 88, 79, 95, 90, 87, 78, 96))
• [Link](3)
Practice Question 5
• A news agency processes article text stored in an RDD, where each
record is a sentence. Compute the word frequency across all
sentences.
Solution
• val textRDD = [Link](Seq("Big data is powerful", "Spark is great
for big data", "Machine learning with Spark"))
• val wordCounts = [Link](_.split(" ")).countByValue()
• wordCounts