Transformations, Actions & Lazy Evaluation

In the chapter "What Apache Spark Is & Why It Exists", Spark did not filter or add up any rows until show(). This chapter explains why. It is one of the most asked Spark ideas in interviews.

You will learn four things. Each one has one or two sections.

1. Every Spark step is a transformation or an action.
2. Spark waits until an action, and this wait helps Spark.
3. Narrow and wide steps decide how Spark splits the work.
4. What happens when you run two actions, or when a step fails.

We use the shop again. The job is the same: total paid revenue for each customer.

The chapter "Spark Architecture: Driver, Executors & Cluster Managers" used a model with 3 partitions of 2 orders each. We use the same model here.

A transformation describes a new DataFrame from an old one. It gives you a DataFrame back. It does not run.

An action asks for a result. It prints rows, counts rows, or writes a file. An action makes Spark run the work.

Here is the whole shop job as 5 steps. Look at the code first.

from pyspark.sql import functions as F

orders = spark.read.parquet("orders.parquet")                          # step 1

paid = orders.filter(F.col("status") == "paid")                        # step 2
with_month = paid.withColumn("order_month", F.month("order_date"))     # step 3
revenue = with_month.groupBy("customer_id").sum("amount")              # step 4

revenue.show()                                                         # step 5

Now put each step into one of the two kinds. The last column uses a word we explain in the next sections.

A quick test: does the call return a DataFrame? Then it is a transformation.

Does it return a value or write output? Then it is an action.

Common actions are show(), count(), collect(), take(), and write. Common transformations are select, filter, withColumn, join, groupBy(...).agg(...), and orderBy.

Spark is lazy. A lazy tool waits and does nothing until it must. When you write steps 1 to 4, Spark only writes them down as a plan.

When you call an action, Spark looks at the whole plan. Then it runs it. Look at what Spark does after each line of the shop job.

Line 1 reads only the small footer of the Parquet file. The footer holds the column names and types. It holds no order rows.

Lines 2 to 4 only add to the plan. Line 5 is the action. Only then does Spark read the 6 orders.

This is also why you can write ten transformations and see nothing happen. You did not make a mistake. Spark is waiting for an action.

The call explain() prints the plan Spark made. It runs no job and shows no rows. Here is the plan for the shop job.

== Physical Plan (simplified) ==
HashAggregate(keys=[customer_id], functions=[sum(amount)])
+- Exchange hashpartitioning(customer_id)
   +- HashAggregate(keys=[customer_id], functions=[partial_sum(amount)])
      +- Project [customer_id, amount]
         +- Filter (status = paid)
            +- FileScan parquet [customer_id, status, amount], PushedFilters: [EqualTo(status,paid)]

Read the plan from the bottom up. Each line is one step, and the lines match the shop job.

• FileScan: Spark reads the file. It lists only 3 columns, because the job needs only those.
• Filter: keep the paid orders.
• Project: keep only the columns the next step needs, customer_id and amount.
• The first HashAggregate: each task adds up its own rows. This is the partial sum.
• Exchange: the shuffle. Partial sums for one customer are sent to one place.
• The last HashAggregate: add the partial sums into the final total.

Look for the column order_month. It is not in the plan. Nothing uses it, so Spark dropped it.

This is called column pruning.

Spark can only do this because it sees the whole plan first. An engine that ran step 3 at once would have built the column for nothing.

Look at PushedFilters on the FileScan line.

Spark passed the filter to the Parquet reader. This is called pushdown. The reader can skip parts of the file that cannot match.

A narrow transformation works on each partition alone. No rows move between machines. A wide transformation needs rows from many partitions.

Spark must move rows. This move is called a shuffle.

Look at the two kinds on the shop data.

Narrow steps next to each other run together inside one task. Each task passes its rows through filter and withColumn in one pass. It does not save results between the two steps.

Here are common steps of each kind.

A shuffle is slow for two reasons. It sends rows over the network, and it writes them to disk first. The shuffle chapter teaches it in depth.

Your steps form a plan. Spark draws the plan as a graph of steps. The name for this graph is a DAG, short for directed acyclic graph.

The steps go one way and never loop back.

When the action runs, Spark cuts the DAG at each wide step. Each piece is a stage. A stage runs one task for each partition.

Our shop job has one wide step, the groupBy. So it has two stages. Look at the three tasks of stage 0.

# Each task runs steps 1 to 3 and a partial sum on its own partition.
revenue.show()

Next comes the shuffle. Spark sends each customer's partial sums to one place.

Here every customer sits in only one partition. Spark does not know that, so it still shuffles.

Stage 1 then adds the partial sums for each customer. This gives the final rows 1: 65.0, 3: 15.5, 4: 80.0. The picture shows the cut.

Add orderBy("customer_id") after the groupBy and you add a second wide step. In a job that writes the result, this means one more shuffle, so one more stage. A later chapter shows how many tasks stage 1 has.

Each action starts at least one new job. Spark does not remember the first result. Look at two actions on the same paid orders.

paid = spark.read.parquet("orders.parquet").filter(F.col("status") == "paid")

paid.count()   # action 1: job 1
paid.show()    # action 2: job 2

The file was read twice. For 6 rows this costs nothing. For a big file on cloud storage, you pay the read two times.

You can keep the result in memory with cache(). Then the second action does not read the file again. The caching chapter in week 11 teaches how to use it.

Laziness changes when you see an error. It depends on the kind of mistake.

First, a wrong column name. Spark checks every name against the table's columns as soon as you run the line. This check is called analysis.

paid = orders.filter(F.col("statuss") == "paid")   # typo: statuss

Second, a bad value. Spark cannot see the values in your rows until the tasks run.

A UDF (user-defined function) is a Python function that you give to Spark. Spark runs it inside the tasks, row by row.

Here is a UDF we wrote on purpose to fail. It divides 100 by the amount minus 40. DoubleType says the result is a decimal number.

from pyspark.sql.types import DoubleType

ratio = F.udf(lambda amount: 100.0 / (amount - 40.0), DoubleType())

bad = paid.withColumn("ratio", ratio("amount"))   # no error here
bad.show()                                         # the error appears here

Spark retried Task 0, as the architecture chapter showed. A bad value fails again on every try, so the job stops.

On a cluster, Spark tries a task 4 times. In local mode it tries only once, so you see "failed 1 times".

Four points to keep.

• Transformations build a plan. Actions run it. Each action starts its own job (or more).
• Spark is lazy so that it can see the whole plan. It can then prune columns and push filters down.
• A wide transformation needs a shuffle, and a shuffle is a stage boundary.
• Errors in names show at once. Errors in values show at the action.

Answer these without looking back.

1. Is groupBy(...).sum(...) a transformation or an action? Is it narrow or wide?
2. In the shop job, which line starts the job?
3. You call count() and then show() on one DataFrame. How many jobs run? How can you avoid the second read?
4. Why does a typo in a column name fail early, but a division by zero fails late?