In Spark, how do you read the physical plan that df.explain() prints?

## The shape of the output `df.explain()` prints a tree with the **root at the top and the leaves at the bottom**, using `+-` for a child and `:-` for the first of two children. Data flows the other way: the bottom-most nodes read files, and each row travels upward through the operators above it. Reading bottom-up is the single habit that makes plans legible. ``` SortMergeJoin [cust_id#12], [cust_id#31], Inner :- Sort [cust_id#12 ASC NULLS FIRST], false, 0 : +- Exchange hashpartitioning(cust_id#12, 200), ENSURE_REQUIREMENTS : +- Filter isnotnull(cust_id#12) : +- FileScan parquet [cust_id#12,amount#14] PushedFilters: [IsNotNull(cust_id)] +- Sort [cust_id#31 ASC NULLS FIRST], false, 0 +- Exchange hashpartitioning(cust_id#31, 200), ENSURE_REQUIREMENTS +- FileScan parquet [cust_id#31,name#32] ``` Here both join inputs are shuffled by `cust_id` and sorted, then merged. The `:` column belongs to the left branch; the `+-` at the same indentation starts the right branch. ## The five things to look for **1. `Exchange` — every shuffle you have.** An `Exchange hashpartitioning(key, 200)` means rows are redistributed by hash of `key` into that many partitions; `ENSURE_REQUIREMENTS` is the preparation rule that inserted it because the operator above demanded that distribution. `Exchange rangepartitioning(...)` comes from a global sort, `Exchange SinglePartition` from an unpartitioned window or a global aggregate — the last one is worth a wince, because it funnels everything through one task. Counting Exchanges tells you how many stage boundaries the query has, which is usually the dominant cost. **2. `*(n)` — whole-stage code generation.** Operators fused into one compiled loop share a codegen stage id, printed as `*(1)`, `*(2)` and so on. Operators without a marker are not fused: a Python UDF evaluation, a source that does not support it, or an operator excluded from codegen. The ids are codegen stages, **not** scheduler stages — do not read `*(3)` as "the third stage in the Spark UI". **3. `PushedFilters` and `ReadSchema` on the scan.** `PushedFilters: [IsNotNull(cust_id), GreaterThan(amount,100.0)]` means the data source itself applies those predicates while reading — for Parquet that includes skipping row groups by their min/max statistics. `ReadSchema: struct<cust_id:string,amount:double>` shows column pruning: only those columns are read off disk. If a filter you wrote is **absent** from `PushedFilters` and instead appears as a separate `Filter` node above the scan, Spark read every row and discarded them in memory. Common reasons: the predicate wraps the column in a function or a cast, it involves a UDF, the source does not support that filter type, or it references two columns of the same table. Partition pruning appears separately as `PartitionFilters`, and that one is the difference between listing a few directories and listing all of them. **4. The join operator's name.** `BroadcastHashJoin` (with a `BroadcastExchange` beneath one side), `SortMergeJoin`, `ShuffledHashJoin`, `BroadcastNestedLoopJoin`, `CartesianProduct`. The last two mean your join condition was not an equality, which is fine for a tiny side and catastrophic otherwise. **5. `AdaptiveSparkPlan isFinalPlan=false` at the top.** With Adaptive Query Execution enabled — the default since Spark 3.2 — what `explain()` prints before execution is the *initial* plan. Spark re-plans after each shuffle using measured statistics, so the operators that actually ran may differ. To see the real one, run the query and open it in the **SQL / DataFrame** tab of the Spark UI, which shows the final plan with runtime metrics per node. ## The explain modes - `explain()` — physical plan only. - `explain(True)` or `explain("extended")` — parsed, analyzed, optimized logical plans plus the physical plan. - `explain("formatted")` — a compact tree of numbered nodes followed by a detail block per node listing inputs, outputs, filter conditions and pushed filters. Best for wide plans, because the tree stops being a wall of nested text. - `explain("cost")` — annotates the optimized logical plan with the statistics the planner used (`sizeInBytes`, `rowCount` when available). This is how you discover that a broadcast decision rested on a wild size estimate. - `explain("codegen")` — prints the generated Java source, which matters when you are chasing a codegen fallback. In SQL, `EXPLAIN FORMATTED SELECT ...` gives the same output. ## A working checklist For a slow query, walk the plan bottom-up and ask: does the scan show my filters and only my columns? How many Exchanges are there, and is any of them `SinglePartition`? Which join algorithm was chosen, and is it what I expected from the input sizes? Where do the `*(n)` markers stop, and why? Then compare with the Spark UI's SQL tab, where the same tree carries actual row counts per node — the gap between the plan's expectation and the measured rows is usually the whole story.