Actions vs Transformations

Spark tutorial · PySpark.in

Transformations and Actions in PySpark

PySpark provides a powerful DataFrame API for processing large data sets in a distributed way. Every operation on a DataFrame falls into one of two categories: transformations or actions. Understanding the difference is essential for writing efficient Spark applications and avoiding common performance pitfalls.

What are Transformations?

A transformation produces a new DataFrame from an existing one. Examples include filtering rows, selecting columns, grouping, joining and sorting. Transformations do not execute immediately. Instead of acting on the data right away, Spark records the steps to be performed and builds a logical execution plan. According to Spark documentation, transformations are lazy: they are not executed until an action is triggered, and they build a plan that Spark later optimizes. Lazy evaluation means Spark can group multiple operations together, reduce the number of data scans and rearrange the order of steps for better performance

Lazy Evaluation:

Transformations are evaluated lazily, meaning they are not executed until an action is triggered.
The execution plan is recorded, and Spark optimizes the plan before executing it.

Let's try to undertand this with a simple example:

Suppose in a code you've 10 steps, out of which first 9 have transformation and last 1 is action.
For the first 9 steps, It'll be lazy evaluation, Spark engine will only provide logical plan and create DAG.
When 10th step is executed, spark engine refers logical plan(DAG) and traverse back to step 1 and will start excuting till step 9.
At the end of step 9, a new dataframe will be created and it will be return to the driver to display.

Examples of Transformations

Transformation	What it does
select()	Select specific columns
filter() / where()	Filter rows based on condition
withColumn()	Create or modify a column
drop()	Remove columns
groupBy()	Group rows for aggregation
join()	Join multiple DataFrames
orderBy()	Sorting rows
dropDuplicates()	Remove duplicate rows

Example of Transformation (No Execution Yet)

1df2 = df.filter(df.salary > 50000)

2print(df2)

▶ Output will appear here.

What are Actions?

Actions are operations that trigger the execution of transformations and return a value to the driver program. They are the operations that actually initiate the computation. Actions are operations that force Spark to execute the logical plan created by transformations and return results to the driver or external storage.

Examples of Actions

Action	What it does
show()	Displays data in tabular format
count()	Returns total number of rows
collect()	Brings all records to driver
first() / head()	Returns first row(s)
take(n)	Returns n rows
toPandas()	Converts to Pandas DataFrame
write()	Write to storage (S3, DB, Delta)

Example of Action (Triggers Execution)

1df2 = df.filter(df.salary > 50000)

2print(df2.take(2))

▶ Output will appear here.

Let's understand Lazy Evaluation and Directed Acyclic Graphs (DAG) with the example depicted in the above picture.

We start with source data, which can be in any format such as CSV or JSON and can originate from any source. This data is stored in DF1 as the source DataFrame.

In step 1, a filter transformation is applied to DF1. Since a filter is a transformation, it doesn't execute immediately. Instead, it creates a logical plan and updates the Directed Acyclic Graph (DAG). At the end of this transformation, a new DataFrame, DF2, is created. Since DataFrames are immutable, all the data inside DF1 remains unchanged, but a new DataFrame is created with filtered records.

Similarly, in step 2, we manipulate the data by concatenating two columns. Since this is also a transformation, it doesn't execute immediately but updates the DAG. At the end, another DataFrame, DF3, is created.

Now, suppose we perform an action by writing DF3 to an Amazon S3 bucket. This action refers to the DAG to check all the transformations and traverses to transformation 1, starting to execute all transformations. It's important to note that not all transformations will necessarily execute at once. The Spark Engine will determine the most cost-effective logical plan to achieve the action. Due to this characteristic, it's considered Lazily evaluated.

There are two kinds of transformations:

1) Narrow Transformation:

Transformations that don't result in data movement between partitions are called Narrow transformations.
Since data in Spark is stored in multiple partitions on multiple machines, these operations are simple and inexpensive, as there is no shuffle.
Functions such as filter() and union() are examples of narrow transformations.

2) Wide Transformation/ Shuffle:

Transformations that involve data movement between partitions are called Wide transformations or shuffle transformations.
Data is shuffled and moved from one partition to another as per the requirement.
These are expensive operations, as in order to complete a task, data needs to be shuffled between partitions.
Because of the above, these operations should be avoided as much as possible.
Functions such as join(), repartition(), and GroupBy() are some examples of wide transformations.

Understanding DAG

The Directed Acyclic Graph (DAG) is a fundamental concept in the context of Apache Spark, representing the logical flow of operations within a Spark job. Understanding the DAG is crucial for comprehending how Spark processes data and optimizing the execution plan. Here’s a breakdown of the key aspects related to the DAG:

1. Logical Execution Plan:

The DAG represents the logical execution plan of a Spark job.
Each node in the DAG corresponds to a transformation or action operation in the Spark job.
Nodes are connected by directed edges, which represent the flow of data between operations.

2. DAG Visualization:

Spark provides tools, such as the Spark UI, for visualizing the DAG of a job.
The Spark UI can be accessed through a web browser and provides detailed information about job progress, resource usage, and the DAG itself.

3. Stages:

The DAG is divided into stages, where each stage represents a set of transformations that can be executed in parallel.
Stages are separated by actions or shuffle operations that require data to be redistributed across the cluster.

4. Tasks:

Tasks are the smallest units of work in Spark and are created based on the partitions of the data.
Each stage consists of a set of tasks that can be executed in parallel on the available worker nodes.

5. Dependencies:

Nodes in the DAG have dependencies on other nodes, reflecting the sequence of operations.
There are two types of dependencies: narrow (one-to-one) and wide (one-to-many).
Narrow dependencies allow for pipelining of data within a single stage, while wide dependencies require shuffling and represent a stage boundary.

6. Optimizations:

Spark performs optimizations on the DAG to improve performance. These optimizations include predicate pushdown, constant folding, and more.
The Catalyst optimizer is responsible for transforming the logical plan into a physical plan, taking into account various optimization techniques.

7. DAG and Lazy Evaluation:

The DAG is constructed during the process of building a logical execution plan.
Transformations are lazy, meaning they are not executed immediately. Instead, they contribute to the DAG.
Actions trigger the actual execution of the DAG, leading to the materialization of results.

8. Performance Tuning:

Understanding the DAG is crucial for performance tuning and optimization.
Analyzing the DAG helps identify potential bottlenecks, unnecessary operations, or opportunities for parallelism.

9. Example:

Suppose you have a sequence of transformations (map, filter, reduceByKey) followed by an action (collect). The DAG would represent these operations and their dependencies, visualizing how data flows through the stages. In summary, the DAG is a visual representation of the logical execution plan of a Spark job. It provides insights into the flow of data, dependencies between operations, and opportunities for optimization, making it a valuable tool for both understanding and improving the performance of Spark applications.