Apache Spark RDD
Apache Spark RDD Transformations and Actions
Apache Spark RDDs support two types of operations:
1. Transformations – Lazy operations that define a new RDD.
2. Actions – Operations that trigger computation and return results.
Transformations
• map(func): Returns a new RDD by applying a function to each element.
• filter(func): Returns a new RDD containing only elements that satisfy the condition.
• flatMap(func): Similar to map, but each input element can produce 0 or more output
elements.
• mapPartitions(func): Runs a function on each partition and returns a new RDD.
• mapPartitionsWithIndex(func): Similar to mapPartitions but provides partition index.
• sample(withReplacement, fraction, seed): Samples a fraction of the data.
• union(otherRDD): Returns the union of two RDDs.
• intersection(otherRDD): Returns the intersection of two RDDs.
• distinct([numTasks]): Returns a new RDD with distinct elements.
• groupByKey(): Groups values with the same key into a list of values.
• reduceByKey(func): Combines values with the same key using the given function.
• aggregateByKey(zeroValue, seqFunc, combFunc): Aggregates values with the same key.
• sortByKey([ascending]): Sorts the RDD by key values.
• sortBy(func): Sorts RDD by computed criteria.
• join(otherRDD): Joins two RDDs based on keys.
• leftOuterJoin(otherRDD): Performs a left outer join on two RDDs.
• rightOuterJoin(otherRDD): Performs a right outer join on two RDDs.
• cogroup(otherRDD): Groups data from both RDDs sharing the same key.
• cartesian(otherRDD): Computes the Cartesian product of two RDDs.
• pipe(command): Pipes elements of an RDD through an external command.
• coalesce(numPartitions): Decreases the number of partitions.
• repartition(numPartitions): Increases or decreases partitions with shuffling.
• repartitionAndSortWithinPartitions(partitioner): Repartitions and sorts within
partitions.
Actions
• collect(): Returns all elements of the RDD as an array.
• count(): Returns the number of elements in the RDD.
• first(): Returns the first element of the RDD.
• take(n): Returns an array with the first n elements.
• takeSample(withReplacement, num, seed): Returns a sample of n elements.
• takeOrdered(n, key=None): Returns the first n elements ordered by a function.
• reduce(func): Reduces the elements using the given function.
• fold(zeroValue, func): Aggregates elements with a neutral zero value.
• aggregate(zeroValue, seqFunc, combFunc): Aggregates using different operations per
partition.
• countByKey(): Counts elements for each key.
• collectAsMap(): Returns key-value pairs as a dictionary.
• lookup(key): Returns all values associated with the key.
• saveAsTextFile(path): Saves RDD as a text file.
• saveAsSequenceFile(path): Saves RDD as a Hadoop sequence file.
• saveAsObjectFile(path): Saves RDD objects in a simple format.
• foreach(func): Applies a function to each element of the RDD.
• foreachPartition(func): Runs a function on each partition of the RDD.
Example 1: Transformations and Actions (Scala)
Scenario
Imagine we have a list of daily sales transactions (amounts in INR). We want to:
1. Square each amount (to simulate some computation).
2. Filter only amounts greater than 1000.
3. Count how many such transactions exist.
4. Collect and print them.
Scala Example
import [Link].{SparkConf, SparkContext}
object RDDExample {
def main(args: Array[String]): Unit = {
// Spark Configuration and Context
val conf = new
SparkConf().setAppName("RDDExample").setMaster("local[*]")
val sc = new SparkContext(conf)
// Sample sales data (transaction amounts in INR)
val data = List(100, 200, 50, 400, 600, 1200, 1500)
val rdd = [Link](data)
// ----------- Transformations -----------
val squared = [Link](x => x * x) // map transformation
val filtered = [Link](x => x > 1000) // filter
transformation
// ----------- Actions -----------
val result = [Link]() // collect action
val count = [Link]() // count action
// Output results
println("Original Data: " + data)
println("Squared Transactions > 1000: " + [Link](", "))
println("Count of Transactions > 1000: " + count)
[Link]()
}
}
Explanation
- map → Transformation → Each transaction is squared.
- filter → Transformation → Keeps only values greater than 1000.
- collect → Action → Retrieves results from the cluster to the driver.
- count → Action → Returns the number of qualifying transactions.
Example Output
Original Data: List(100, 200, 50, 400, 600, 1200, 1500)
Squared Transactions > 1000: 160000, 360000, 1440000, 2250000
Count of Transactions > 1000: 4
Example 2: Transformations and Actions (Scala)
1. Split sentences into words.
2. Remove duplicates.
3. Count word frequencies.
4. Sort words alphabetically.
5. Display top results.
object WordCountExample {
def main(args: Array[String]): Unit =
{
// Spark Configuration and Context
val conf = new SparkConf().setAppName("WordCountExample").setMaster("local[*]")
val sc = new SparkContext(conf)
// Sample dataset (log messages or sentences)
val sentences = List(
"Apache Spark is fast",
"Spark runs on clusters",
"Apache Spark is powerful"
)
val rdd = [Link](sentences)
// ----------- Transformations -----------
val words = [Link](line => [Link](" ")) // flatMap → split into words
val uniqueWords = [Link]() // distinct → remove duplicates
val pairs = [Link](word => (word, 1)) // map → create (word, 1) pairs
val wordCounts = [Link](_ + _) // reduceByKey → count occurrences
val sorted = [Link]() // sortByKey → sort alphabetically
// ----------- Actions -----------
val allWords = [Link]() // collect
val counts = [Link]() // collect results
val totalWords = [Link]() // count
// Output results
println("Unique Words: " + [Link](", "))
println("Word Counts: ")
[Link](println)
println("Total Words: " + totalWords)
[Link]()
}
}
flatMap → splits each sentence into words.
distinct → ensures unique words.
map → maps each word to (word,1).
reduceByKey → groups by word and counts.
sortByKey → sorts alphabetically.
collect & count → Actions to bring results to driver.
Unique Words: Apache, Spark, is, fast, runs, on, clusters, powerful
Word Counts:
(Apache,2)
(Spark,3)
(clusters,1)
(fast,1)
(is,2)
(on,1)
(powerful,1)
(runs,1)
Total Words: 11
Example 3: Transformations and Actions (Scala)
We have employee salary data with (department, salary) pairs.
1. Group employees by department (groupByKey).
2. Find the average salary per department (aggregateByKey).
3. Print which partition holds which records (mapPartitionsWithIndex).
4. Reduce the number of partitions (coalesce).
5. Repartition and sort data within partitions (repartitionAndSortWithinPartitions)
import [Link].{SparkConf, SparkContext}
import [Link]
object EmployeeSalaryExample
{
def main(args: Array[String]): Unit =
{
val conf = new
SparkConf().setAppName("EmployeeSalaryExample").setMaster("local[*]")
val sc = new SparkContext(conf)
// Employee data: (Department, Salary)
val data = List(
("HR", 30000), ("IT", 50000), ("IT", 60000),
("HR", 40000), ("Finance", 45000), ("Finance", 55000),
("IT", 70000)
)
val rdd = [Link](data, 3)
// ----------- Transformations -----------
val grouped = [Link]() // Group salaries by department
val avgSalary = [Link]((0,0))(
(acc, salary) => (acc._1 + salary, acc._2 + 1), // Within partition (sum, count)
(acc1, acc2) => (acc1._1 + acc2._1, acc1._2 + acc2._2) // Across partitions
).mapValues{ case (sum, count) => sum / count } // Compute average
val partitionInfo = [Link](
(index, iter) => [Link](x => s"Partition $index contains $x")
)
val reducedPartitions = [Link](2) // Reduce partitions
val repartitioned = [Link](new HashPartitioner(2))
// ----------- Actions -----------
println("Grouped by Department: " + [Link]().mkString(", "))
println("Average Salary by Department: " + [Link]().mkString(", "))
println("Partition Info: " + [Link]().mkString(" | "))
println("Coalesced Partitions: " + [Link]().mkString(", "))
println("Repartitioned & Sorted: " + [Link]().mkString(", "))
[Link]()
}
}
groupByKey → Groups all salaries under each department.
aggregateByKey → Computes average salary per department efficiently.
mapPartitionsWithIndex → Shows how data is distributed across partitions.
coalesce → Reduces partitions without shuffle (useful for optimization).
repartitionAndSortWithinPartitions → Rebalances data and sorts it within each
partition.