RDD Actions
RDD actions are operations that return the raw values. In other words, any RDD function that returns other than JavaRDD<T> is considered an action in Spark programming. In this lesson, we will learn RDD actions with Java Spark examples.
Action functions trigger the transformations to execute. As mentioned in RDD Transformations, all transformations are lazy, meaning they do not get executed right away, and to execute the transformations, you need to call an action.
The following are the most used RDD actions. For the complete list, please refer to Spark's official documentation.
| RDD ACTION METHODS | METHOD DEFINITION |
| ------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| aggregate() | Aggregate the elements of each partition, and then the results for all the partitions. |
| collect() | Return the complete dataset as an Array. |
| count() | Return the count of elements in the dataset. |
| countByValue() | Return Map<T, Long> key representing each unique value in dataset and value represent count for each value present. |
| first() | Return the first element in the dataset. |
| fold() | Aggregate the elements of each partition, and then the results for all the partitions. |
| foreach() | Iterates all elements in the dataset by applying function f to all elements. |
| foreachPartition() | Similar to foreach, but applies function f for each partition. |
| max() | Return the maximum value from the dataset. |
| min() | Return the minimum value from the dataset. |
| reduce() | Reduces the elements of the dataset using the specified binary operator. |
| saveAsObjectFile() | Saves RDD as a serialized object's to the storage system. |
| saveAsTextFile() | Saves RDD as a text file. |
| take(num: Int) | Return the first num elements of the dataset. |
| takeSample() | Return the subset of the dataset in an Array. Note: Use this method only when the resulting array is small, as all the data is loaded into the driver's memory. |
| top(num: Int) | Note: Use this method only when the resulting array is small, as all the data is loaded into the driver's memory. |
| treeAggregate | Aggregates the elements of this RDD in a multi-level tree pattern. |
| treeReduce | Reduces the elements of this RDD in a multi-level tree pattern. |
RDD Actions Example
Before we start explaining RDD actions with examples, first, let's create an RDD. Note that we have created two RDDs in the below code snippet and we use these two as needed to demonstrate the RDD actions.
// Create RDD from a List Using parallelize
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import java.util.Arrays;
import java.util.List;
public class RDDExample {
public static void main(String[] args) {
SparkSession spark = SparkSession.builder()
.appName("SparkByExample")
.master("local")
.getOrCreate();
// Create RDD
JavaSparkContext sc = new JavaSparkContext(spark.sparkContext());
List<Tuple2<String, Integer>> data = Arrays.asList(
new Tuple2<>("Z", 1), new Tuple2<>("A", 20),
new Tuple2<>("B", 30), new Tuple2<>("C", 40),
new Tuple2<>("B", 30), new Tuple2<>("B", 60));
JavaRDD<Tuple2<String, Integer>> inputRDD = sc.parallelize(data);
JavaRDD<Integer> listRdd = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5, 3, 2));
}
}
aggregate()
aggregate() the elements of each partition, and then the results for all the partitions.
import scala.Tuple2;
//aggregate
Function2<Integer, Integer, Integer> param0 = Integer::sum;
Function2<Integer, Integer, Integer> param1 = Integer::sum;
System.out.println("aggregate : " + listRdd.aggregate(0, param0, param1));
//Output: aggregate : 20
Function2<Integer, Tuple2<String, Integer>, Integer> param3 = (accu, v) -> accu + v._2;
Function2<Integer, Integer, Integer> param4 = Integer::sum;
System.out.println("aggregate : " + inputRDD.aggregate(0, param3, param4));
//Output: aggregate : 181
treeAggregate()
treeAggregate() – Aggregates the elements of this RDD in a multi-level tree pattern. The output of this function will be similar to the aggregate function.
//treeAggregate. This is similar to aggregate
Function2<Integer, Integer, Integer> param8 = Integer::sum;
Function2<Integer, Integer, Integer> param9 = Integer::sum;
System.out.println("treeAggregate : " + listRdd.treeAggregate(0, param8, param9));
//Output: treeAggregate : 20
fold()
fold() – Aggregate the elements of each partition, and then the results for all the partitions.
//fold
Function2<Integer, Integer, Integer> seqOp = Integer::sum;
System.out.println("fold : " + listRdd.fold(0, seqOp));
//Output: fold : 20
Function2<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> foldFunc = (acc, v) -> new Tuple2<>("Total", acc._2 + v._2);
System.out.println("fold : " + inputRDD.fold(new Tuple2<>("Total", 0), foldFunc));
//Output: fold : (Total,181)
reduce
reduce() – Reduces the elements of the dataset using the specified binary operator.
//reduce
System.out.println("reduce : " + listRdd.reduce(Integer::sum));
//Output: reduce : 20
System.out.println("reduce alternate : " + listRdd.reduce((x, y) -> x + y));
//Output: reduce alternate : 20
Function2<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple2<String, Integer>> reduceFunc = (x, y) -> new Tuple2<>("Total", x._2 + y._2);
System.out.println("reduce : " + inputRDD.reduce(reduceFunc));
//Output: reduce : (Total,181)
treeReduce
treeReduce() – Reduces the elements of this RDD in a multi-level tree pattern.
//treeReduce. This is similar to reduce
System.out.println("treeReduce : " + listRdd.treeReduce(Integer::sum));
//Output: treeReduce : 20
collect
collect() - Return the complete dataset as an Array.
//Collect
int[] dataArray = listRdd.collect().stream().mapToInt(i -> i).toArray();
for (int i : dataArray) {
System.out.println(i);
}
count, countApprox, countApproxDistinct
count() – Return the count of elements in the dataset.
countApprox() – Return approximate count of elements in the dataset, this method returns incomplete when execution time meets timeout.
countApproxDistinct() – Return an approximate number of distinct elements in the dataset.
//count, countApprox, countApproxDistinct
System.out.println("Count : " + listRdd.count());
//Output: Count : 7
System.out.println("countApproxDistinct : " + listRdd.countApproxDistinct());
//Output: countApproxDistinct : 5
System.out.println("countApproxDistinct : " + inputRDD.countApproxDistinct());
//Output: countApproxDistinct : 5
countByValue, countByValueApprox
countByValue() – Return Map<T, Long> key representing each unique value in dataset and value represents count for each value present.
countByValueApprox() – Same as countByValue() but returns approximate result.
//countByValue, countByValueApprox
System.out.println("countByValue : " + listRdd.countByValue());
//Output: countByValue : {5=1, 1=1, 2=2, 3=2, 4=1}
first
first() – Return the first element in the dataset.
//first
System.out.println("first : " + listRdd.first());
//Output: first : 1
System.out.println("first : " + inputRDD.first());
//Output: first : (Z,1)
top
top() – Return top n elements from the dataset.
Note: Use this method only when the resulting array is small, as all the data is loaded into the driver’s memory.
//top
System.out.println("top : " + listRdd.top(2));
//Output: top : [5, 4]
System.out.println("top : " + inputRDD.top(2));
//Output: top : [(C,40), (B,60)]
min
min() – Return the minimum value from the dataset.
//min
System.out.println("min : " + listRdd.min());
//Output: min : 1
System.out.println("min : " + inputRDD.min());
//Output: min : (A,20)
max
max() – Return the maximum value from the dataset.
//max
System.out.println("max : " + listRdd.max());
//Output: max : 5
System.out.println("max : " + inputRDD.max());
//Output: max : (Z,1)
take, takeOrdered, takeSample
take() – Return the first num elements of the dataset.
takeOrdered() – Return the first num (smallest) elements from the dataset and this is the opposite of the take() action.
Note: Use this method only when the resulting array is small, as all the data is loaded into the driver’s memory.
takeSample() – Return the subset of the dataset in an Array.
Note: Use this method only when the resulting array is small, as all the data is loaded into the driver’s memory.
//take, takeOrdered, takeSample
System.out.println("take : " + listRdd.take(2));
//Output: take : [1, 2]
System.out.println("takeOrdered : " + listRdd.takeOrdered(2));
//Output: takeOrdered : [1, 2]
In Summary, RDD actions are operations that return non-RDD values. Since RDDs are lazy, they do not execute the transformation functions until we call actions. Hence, all these functions trigger the transformations to execute and finally return the value of the action functions to the driver program. In this tutorial, you have also learned several RDD function usages and examples.
Further Reading
For additional learning on this topic, I would recommend reading the following from AIStudyLabs.com.