RDD Actions
RDD actions are operations that return the raw values. In other words, any RDD function that returns something other than RDD[T] is considered an action in PySpark programming. In this lesson, we will learn RDD actions with PySpark 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. 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 PySpark'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 a list. | | count() | Return the count of elements in the dataset. | | countByValue() | Return a map with each unique value in the dataset as a key and its count as the value. | | first() | Return the first element in the dataset. | | fold() | Aggregate the elements of each partition, and then the results for all the partitions. | | foreach() | Apply a function to each element in the dataset. | | foreachPartition() | Similar to foreach, but applies a function to each partition. | | max() | Return the maximum value from the dataset. | | min() | Return the minimum value from the dataset. | | reduce() | Reduce the elements of the dataset using the specified binary operator. | | saveAsObjectFile() | Save RDD as a serialized object to the storage system. | | saveAsTextFile() | Save RDD as a text file. | | take(num: Int) | Return the first num elements of the dataset. | | takeSample() | Return a subset of the dataset in a list. Note: Use this method only when the resulting list is small, as all the data is loaded into the driver's memory. | | top(num: Int) | Note: Use this method only when the resulting list is small, as all the data is loaded into the driver's memory. | | treeAggregate | Aggregate the elements of this RDD in a multi-level tree pattern. | | treeReduce | Reduce 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 code snippet below, and we will use these two as necessary to demonstrate the RDD actions.
# Create RDD from a List Using parallelize
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("PySparkByExample") \
.master("local") \
.getOrCreate()
# Create RDD
inputRDD = spark.sparkContext.parallelize([("Z", 1), ("A", 20), ("B", 30), ("C", 40), ("B", 30), ("B", 60)])
listRdd = spark.sparkContext.parallelize([1, 2, 3, 4, 5, 3, 2])
aggregate()
aggregate() the elements of each partition, and then the results for all the partitions.
# aggregate
def param0(accu, v):
return accu + v
def param1(accu1, accu2):
return accu1 + accu2
print("aggregate :", listRdd.aggregate(0, param0, param1))
# Output: aggregate : 20
def param3(accu, v):
return accu + v[1]
print("aggregate :", inputRDD.aggregate(0, param3, param1))
# 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
print("treeAggregate :", listRdd.treeAggregate(0, param0, param1))
# Output: treeAggregate : 20
fold()
fold() – Aggregate the elements of each partition, and then the results for all the partitions.
# fold
print("fold :", listRdd.fold(0, lambda acc, v: acc + v))
# Output: fold : 20
print("fold :", inputRDD.fold(("Total", 0), lambda acc, v: ("Total", acc[1] + v[1])))
# Output: fold : ('Total', 181)
reduce
reduce() – Reduces the elements of the dataset using the specified binary operator.
# reduce
print("reduce :", listRdd.reduce(lambda x, y: x + y))
# Output: reduce : 20
print("reduce :", inputRDD.reduce(lambda x, y: ("Total", x[1] + y[1])))
# Output: reduce : ('Total', 181)
treeReduce
treeReduce() – Reduces the elements of this RDD in a multi-level tree pattern.
# treeReduce. This is similar to reduce
print("treeReduce :", listRdd.treeReduce(lambda x, y: x + y))
# Output: treeReduce : 20
collect
collect() - Return the complete dataset as a list.
# Collect
data = listRdd.collect()
for elem in data:
print(elem)
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
print("Count :", listRdd.count())
# Output: Count : 7
# countApprox is not available in PySpark
print("countApproxDistinct :", listRdd.countApproxDistinct())
# Output: countApproxDistinct : 5
print("countApproxDistinct :", inputRDD.countApproxDistinct())
# Output: countApproxDistinct : 4
countByValue, countByValueApprox
countByValue() – Return a map with each unique value in the dataset as a key and its count as the value.
countByValueApprox() – Same as countByValue() but returns approximate result.
# countByValue, countByValueApprox
print("countByValue :", listRdd.countByValue())
# Output: countByValue : defaultdict(<class 'int'>, {1: 1, 2: 2, 3: 2, 4: 1, 5: 1})
first
first() – Return the first element in the dataset.
# first
print("first :", listRdd.first())
# Output: first : 1
print("first :", inputRDD.first())
# Output: first : ('Z', 1)
top
top() – Return top n elements from the dataset.
Note: Use this method only when the resulting list is small, as all the data is loaded into the driver’s memory.
# top
print("top :", listRdd.top(2))
# Output: top : [5, 4]
print("top :", inputRDD.top(2))
# Output: top : [('Z', 1), ('C', 40)]
min
min() – Return the minimum value from the dataset.
# min
print("min :", listRdd.min())
# Output: min : 1
print("min :", inputRDD.min())
# Output: min : ('A', 20)
max
max() – Return the maximum value from the dataset.
# max
print("max :", listRdd.max())
# Output: max : 5
print("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 list is small, as all the data is loaded into the driver’s memory.
takeSample() – Return a subset of the dataset in a list. Note: Use this method only when the resulting list is small, as all the data is loaded into the driver’s memory.
# take, takeOrdered, takeSample
print("take :", listRdd.take(2))
# Output: take : [1, 2]
print("takeOrdered :", listRdd.takeOrdered(2))
# Output: takeOrdered : [1, 2]
print("takeSample :", listRdd.takeSample(False, 2))
# Example output: takeSample : [2, 3]
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 functions' usage and examples.
Further Reading
For additional learning on this topic, I would recommend reading the following from AIStudyLabs.com.