The Daily Grind's busiest-hour report takes forty minutes, and the code is eleven lines. Which line? You cannot tell from the code, because Spark does not run your code. It runs a plan it made from your code, and the plan is where the forty minutes are. The Where Your Code Runs lesson printed one plan and read it for the shuffle. This lesson is Part 2's opener because the next three lessons all start from a plan: it reads all four that explain can print.
Watch it happen
The same eleven Daily Grind orders, grouped by size:
summary = orders.groupBy("size").agg(F.count("*").alias("n"))
summary.explain("extended")
result = summary
| size | n |
|---|---|
| large | 5 |
| small | 4 |
| medium | 2 |
With "extended", four plans print to the Console tab, top to bottom:
== Parsed Logical Plan == what you wrote, as a tree
== Analyzed Logical Plan == names and types resolved
== Optimized Logical Plan == what Spark decided to do instead
== Physical Plan == how it will be done: the Exec lines
The interesting one is the third. In the parsed plan, the scan is followed by a projection that carries every column of the table:
Projection: ?table?.order_id AS #0, ?table?.drink AS #1, ?table?.size AS #2, ?table?.price AS #3, ?table?.tip AS #4
TableScan: ?table?
In the optimised plan, the scan itself reads one:
Projection: ?table?.size AS #2
TableScan: ?table? projection=[size]
(?table? is Sail's name for a table built in memory; #2 is its number for the size column.) Nobody asked for that. The optimiser noticed that n needs no column and size needs one, and cut the other four from the read. That is column pruning, and it happened to code that never mentioned it.
What the optimiser is allowed to do
Spark's optimiser, Catalyst, rewrites the logical plan under one rule: the result must be the same. Under that rule it will prune columns, push a filter down towards the scan so fewer rows are carried, fold constants, and merge two filters into one. Watch the merge:
q = (orders.filter(F.col("size") == "large")
.filter(F.col("price") > 5)
.select("order_id"))
q.explain("extended")
result = q
| order_id |
|---|
| 1001 |
| 1003 |
| 1005 |
| 1007 |
| 1010 |
Parsed: two Filter lines, one above the other, in the order you wrote them. Optimised: one.
Projection: ?table?.order_id AS #5
Filter: ?table?.size = Utf8("large") AND ?table?.price > Float64(5)
TableScan: ?table? projection=[order_id, size, price]
The scan reads the three columns the filter and the select need, and the two filters are one predicate. The A Date Is Not a String lesson met this merge the hard way: a .filter(pattern) before a to_date and a filter after it became one predicate, so the parse ran on the rows the first filter was meant to exclude. Not a bug: the optimiser doing what it is for. The fix was F.when inside the expression, where no rewrite can move it.
Reading the physical plan
Bottom up, as before. Three things to look for:
RepartitionExechere,Exchangeon Spark: a shuffle. Count them; stages are shuffles plus one (Who Does What on a Cluster).- The join operator's name.
HashJoinExecorNestedLoopJoinExechere;BroadcastHashJoin,SortMergeJoinorBroadcastNestedLoopJoinon Spark. How a Join Is Planned says what each costs. partitions=on the scan, and the number afterHash(...)on a shuffle: how many tasks each stage will be. How Many Pieces is next.
explain("formatted") and explain("cost") are two more layouts of the same plans, the second with row and size statistics where the engine has them; Sail accepts both, in its own layout.
What this site shows and what it can't
Sail prints the same four plans, with its own operator names, so everything above was measured here. What you cannot see here is the plan with numbers on it: on a cluster the Spark UI's SQL tab draws this physical plan with rows read and bytes shuffled on every operator, which is how the forty minutes get a line number.
The symptom, and the fix
A job is slow and the code is short. Print explain("extended"), read the optimised plan for what Spark decided, then the physical plan for the shuffles. The forty minutes are almost always on a shuffle line, and the fix is in the lessons that follow: fewer or more pieces, a less skewed key, a broadcast.
Self-check
A plan's optimised section reads TableScan projection=[size] for a query whose code selects three columns. Is something wrong?
No: column pruning. The query's result needs one column, so the optimiser read one. The result is the same, which is the only rule it has to keep.
Nothing to grade here. Next, How Many Pieces: the number on the shuffle line is the number of tasks, and two methods change it.