Understanding Spark Transformations and Actions Through Source Code

RDD is the core abstraction for working with distributed data in Apache Spark. We process data by applying operations such as transformations to RDD elements.

Note: as of writing, DataFrame is more commonly used than RDD in production systems.

For creating RDDs, PySpark provides two main ways:

• loading an external dataset
• distributing an in-memory collection

Example:

from pyspark.core.context import SparkContext

sc = SparkContext(
    master="local[*]",
    appName="rdd-test"
)

# Reduce logs
sc.setLogLevel("ERROR")

lines = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11])

RDD provide two types of operations

• transformations
• actions

Transformations construct a new RDD from an existing one.

On the other hand, actions compute a result based on an RDD either:

• returning a result to the driver program
• or writing data to external storage.

Spark handle them differently because transformations are lazily evaluated.

In other words, transformations are not executed immediately. Spark records the transformations internally, and actual execution starts only when an action is invoked.

For example:

rdd.filter(lambda x: x % 2 == 0)

does not immediately process the data.

Spark starts execution only when an actiojns such as collect() or count() is called.

Another important point is that transformations create new RDDs instead of modifying existing ones.

The new RDD internally contains information such as:

• parent RDD
• transformation function
• partition information

Diving into the Source Code

internally, filter() looks like this:

def filter(self: "RDD[T]", f: Callable[[T], bool]) -> "RDD[T]":
        def func(iterator: Iterable[T]) -> Iterable[T]:
            return filter(fail_on_stopiteration(f), iterator)

        return self.mapPartitions(func, True)

As you can see, filter method internally calls mapPartitions(). Looking further into the source code, mapPartitions() eventually creates a PipelinedRDD.

According to the source code, it acsts as a container that stores lineage information such as:

• previous RDD
• transformation functions
• dependencies between RDDs

Spark uses this lineage information later when an action is executed.

What Happened When an Action Is Invoked?

The interesting part is that PySpark interacts with the Apache Spark engine written in Scala.

For example, collect() internally does this:

    def collect(self: "RDD[T]") -> List[T]:
        with SCCallSiteSync(self.context):
            assert self.ctx._jvm is not None
            sock_info = self.ctx._jvm.PythonRDD.collectAndServe(self._jrdd.rdd())
        return list(_load_from_socket(sock_info, self._jrdd_deserializer))

PySpark communicates with the JVM Spark engine through Py4J. I will state that more deeply in other blog.

So, Spark executes the transformations only after an action is invoked, processing the lineage form the beginning to the final operation.