---
name: optimize-spark-sql-plan
description: Optimize Apache Spark SQL and DataFrame queries using the final Adaptive Query Execution plan and runtime statistics rather than source code alone. Use to reduce runtime, shuffle, spill, scan cost, skew, join amplification, Python UDF overhead, poor partitioning, or unnecessary work while preserving query semantics.
---

# Optimize a Spark SQL plan

Find the highest-impact improvement supported by the executed plan and its runtime metrics. Work from the final adaptive plan (`isFinalPlan=true`), not source code or the initial plan: AQE may already have broadcast, coalesced, or split what you were about to recommend.

## Get the evidence

```bash
export SPARK_HISTORY_URL="https://history.example.com"
python3 scripts/spark_history_api.py sql-list --app-id <application-id>
python3 scripts/spark_history_api.py sql --app-id <application-id> --execution-id <execution-id> > /tmp/spark-sql.json
```

Run from this skill directory. If `SPARK_HISTORY_URL` is unset, find the server before asking the user: try `http://localhost:18080`, a running application's UI on `http://localhost:4040`, and the history-server or eventLog settings in the local Spark config; ask only when nothing responds. Authentication comes from `SPARK_HISTORY_AUTHORIZATION`, `SPARK_HISTORY_COOKIE`, or `SPARK_HISTORY_HEADERS_JSON`; never ask for credentials in chat and never disable TLS verification (`--ca-file` for a private CA).

## Collector reference

`scripts/spark_history_api.py` subcommands (read-only; each also accepts `--base-url`, `--stage-limit N`, `--task-limit N`):

- `applications [--status completed|running] [--limit N]`: list applications and attempt IDs.
- `failure --app-id <id>`: `failedJobs`, `failedStages[]` (stage, `taskSummary` quantiles, failed-task samples), `allExecutors`, `environment`, `sqlExecutions`.
- `slow --app-id <id>`: `jobs`, `longestStages[]` (stage, `taskSummary` quantiles, slowest-task samples), `allExecutors`, `environment`, `sqlExecutions`.
- `sql-list --app-id <id>`: SQL executions with IDs, status, and duration.
- `sql --app-id <id> --execution-id <n>`: `sqlExecution` (plan text plus per-node metrics), `jobs`, `longestStages`, `environment`.

For anything the profiles omit, call the same REST API directly at `$SPARK_HISTORY_URL/api/v1` with the auth headers above: `/applications/{app}/jobs`, `/stages/{stage}/{attempt}/taskSummary?quantiles=0.05,0.5,0.95`, `/stages/{stage}/{attempt}/taskList?sortBy=-runtime`, `/allexecutors`, `/environment`, `/sql/{execution}?details=true&planDescription=true`, and `/logs` (full event-log zip; large).

The snapshot's layout: `sqlExecution.planDescription` is the executed plan text (find the `isFinalPlan=true` sections), `sqlExecution.nodes[].metrics` carry per-operator counters ("number of output rows", "data size", "spill size", files and bytes read) with `edges` giving parent-child links, `jobs[].stageIds` tie the execution to stages, `longestStages[]` holds each stage plus its `taskSummary` quantiles, and `environment.sparkProperties` is the effective configuration. Slice it rather than reading it whole:

```bash
jq '.sqlExecution.nodes[] | select(.nodeName | test("Join")) | {nodeName, metrics}' /tmp/spark-sql.json
```

If the API omits the plan, recover it from the driver log or generate `explain(mode="formatted")` from the deployed code yourself. Get the deployed query code too: the plan tells you what ran, the code tells you what was intended.

## Highest-impact opportunities, in order, and where to look

1. **Cardinality blowup.** For each join in `sqlExecution.nodes`, compare its "number of output rows" against its children's (via `edges`). Output far exceeding both inputs is usually an unintended many-to-many: fix keys or deduplicate first, because it dominates every downstream metric.
2. **Pruning.** Scan nodes' metrics ("number of files read", bytes read, output rows) against what the query uses, and the pushed/partition filters shown on the scan in `planDescription`. Missing partition filters, unpushed predicates, and wide reads for narrow queries show up here.
3. **Broadcast.** A `SortMergeJoin` whose smaller side's observed "data size" metric is broadcastable when AQE did not already convert it; stale statistics are the usual reason it missed. Check `spark.sql.autoBroadcastJoinThreshold` in `environment.sparkProperties` and executor memory before recommending it.
4. **Unnecessary shuffles.** Each `Exchange` and its partitioning expression in `planDescription`: exchanges whose partitioning an upstream operation already satisfies, back-to-back exchange/sort pairs, or repartitions the query does not need. Every exchange should map to a semantic requirement.
5. **Skewed joins and aggregations.** Locate the join's stage through `jobs[].stageIds`, then compare max against median shuffle read and run time in that stage's `taskSummary` under `longestStages` (re-run with a higher `--stage-limit` if it is not there). Check `planDescription` for AQE skew handling (`skewed=true`) and, if absent, why.
6. **Partition count.** `AQEShuffleRead` coalescing in `planDescription`, stage `numTasks` in `longestStages[].stage` against cores, and spill in `taskSummary`. For uniformly oversized shuffle partitions, raise `spark.sql.shuffle.partitions` or AQE targets before adding explicit repartitions, which can insert redundant exchanges.
7. **Excess spill in stateful operators.** "Spill size" on `HashAggregate`, `Sort`, and `Window` nodes in `sqlExecution.nodes`: reduce state (partial aggregation, narrower rows, fewer window columns) before adding memory.
8. **Python UDF boundaries.** `BatchEvalPython`/`ArrowEvalPython` in `sqlExecution.nodes` and their time metrics relative to rows: a native expression often removes the serialization boundary, which costs more than the function itself.
9. **Caching.** Count `InMemoryTableScan` nodes: a cached result scanned once wastes memory, and a subtree appearing recomputed under several plan branches wants a cache.

An `Exchange` is not automatically waste, a sort-merge join is not automatically wrong, a broadcast is not automatically safe, and a Python UDF is not automatically material: tie every recommendation to the node's runtime metrics, and check whether Catalyst or AQE already applied it.

Changes must preserve semantics: row counts, join cardinality, null behavior, duplicates, and ordering assumptions. Changing partitioning can alter low-order bits of floating-point aggregations.

## Report

State the highest-impact change with its plan node, runtime evidence, and expected effect; why Spark did not already do it; correctness risks; and any remaining smaller findings. Do not apply hints, code, or configuration changes to production without approval.