Diagnosing a slow transformation

Lesson 5 of 5 in Transformation Engines.

There are four things a slow job can be short of, and they have different cures. Diagnosing in the wrong order is how a team spends a sprint on the wrong one.

It is not short of anything — it is serial. One task reading one non-splittable object, or a stage with one task because the input is one file.
It is short of parallelism. The work divides cleanly and there are not enough cores. This is the only case where "add workers" is the answer, and it is the least common.
It is moving data it did not need to move. A shuffle amplified by filtering too late, a sort-merge join where a broadcast would do, or a window function over a high-cardinality key.
It is spending its time on per-file overhead rather than on bytes. The small-files problem, where listing, opening and scheduling exceed reading.

Four symptoms distinguish them, and all four are readable from the Spark UI or the Glue job metrics without changing a line of code. One long task while everything idles is (1) or skew. Every task short and the stage long is (4) — too many tasks. Shuffle Read far above the input size is (3). All cores busy for the whole run is (2), and only then does a bigger cluster help.

Two things in that simulation deserve emphasis because they are where the exam puts its questions.

The small-files failure lands on the driver, not the executors. The driver builds the in-memory file index while listing S3, and with enough paths it exhausts its heap — java.lang.OutOfMemoryError: Java heap space on a job whose data is tiny. AWS documents roughly 650,000 files as the practical ceiling for a standard worker without help. Scaling out does nothing, because the driver is one process and is not one of the executors. The levers that do work: a larger worker type (more driver memory), useS3ListImplementation to reduce listing memory, bounded execution with boundedFiles or boundedSize to process a slice per run with bookmarks carrying the position, or fewer files.

File grouping has a format limitation that is exactly backwards from what you want. groupFiles: "inPartition" with a groupSize puts many files into one in-memory partition and one task, and Glue enables grouping automatically above 50,000 input files. But it is supported for CSV, JSON, ION, XML and grokLog — and not for Parquet, ORC or Avro. So the columnar formats, where small files hurt most because of per-file footer reads, are the ones with no read-time escape hatch. For them the answer is compaction.

Compaction is the cure rather than the coping strategy: read a partition and rewrite it as a handful of right-sized objects, targeting roughly 128 MB to 1 GB each. In Spark, coalesce(n) reduces partitions without a shuffle by merging adjacent ones and is the cheap choice when reducing; repartition(n) shuffles to get an even distribution and is what you need when increasing the count or when the existing partitions are skewed.

And the interaction that only shows up in the bill: rewriting files changes their last-modified times, which is exactly what an S3 job bookmark keys on — so a downstream incremental job will reprocess everything the compaction touched.

The job is slow. What is it short of?

Interactive decision tree — outcomes:

Serial input — a splittability problem
One task means one indivisible input. Almost always a single large gzip object, occasionally a single-file read of any kind. Nothing about cluster size matters: the input cannot be divided.

Fixes, in order: ask the producer for many moderate files; re-encode once into Parquet with Snappy so the container splits; or use bzip2 if a single large text file must stay text. Adding cores is the one thing that definitely does not help.
[[data-skew|Data skew]] — one task is the stage
The stage ends when its last task does, so the maximum is the only number that matters — and the median often improves as skew worsens, which is why average and total CPU metrics conceal it completely.

In order of cost: filter out nulls and sentinel values before the join (free, and often the whole fix); enable spark.sql.adaptive.skewJoin.enabled so AQE splits the outlier and replicates the matching side; broadcast the small side to remove the exchange; or salt the key manually. More executors changes nothing, because one task runs on one core.
[[small-files-problem|Per-file overhead]] — the data is not the problem
Thousands of millisecond tasks means the task count is the file count. Listing, opening and scheduling now exceed reading, and if the file count is large enough the driver runs out of heap building its index — roughly 650,000 files on a standard worker.

Fixes: groupFiles: "inPartition" with a groupSize (CSV, JSON, ION, XML and grokLog only — not Parquet, ORC or Avro); boundedFiles with bookmarks to process a slice per run; a larger worker for driver memory; and compaction as the actual cure. If the producer is a streaming sink, widen its buffering hints so the files are never written small.
The [[spark-shuffle|exchange]] is the cost
Shuffle Read above the input size means the wide transformation amplified the data: every row serialised, written to local disk, fetched over the network and re-read, with writers x readers connections.

Fixes: filter and project before the wide operation; replace a sort-merge join with a broadcast where one side is small (and use an explicit hint if the statistics over S3 are wrong); and let AQE coalesce the shuffle partitions by not pinning spark.sql.shuffle.partitions, since setting it explicitly disables that coalescing. Adding executors makes the exchange wider, not faster.
Genuinely short of parallelism
Every core busy for the whole run, tasks of similar length, and the shuffle is not dominating. This is the case where more workers helps, and it is the least common of the four — which is why it is worth reaching last rather than first.

Choose the lever that matches: more workers for parallelism, a bigger worker type for memory and disk per task. Then consider Flex execution if nothing downstream has a deadline, and Glue auto scaling so you specify a maximum rather than a fixed count and stop paying for idle executors during the narrow stages.

Read the plan, not the tea leaves

For a slow SQL transformation, the execution plan names the dominant operation in a way guessing cannot. What to look for, in the order that pays off:

The Exchange nodes. Each one is a shuffle, and their number is the first-order cost of the job. df.explain(True) or the Spark UI's SQL tab shows them.

The join strategy. A broadcast hash join against a small side is cheap; a sort-merge join against two large sides is the expensive node. In Redshift the equivalent markers are DS_BCAST_INNER (broadcasting the inner table), DS_DIST_BOTH (redistributing both — the expensive one) and DS_DIST_NONE (already co-located by distribution key, so nothing moved).

What was pushed down. A plan showing the partition filter and the column list at the scan node is pruning; one showing a full scan with the filter applied above it is not.

Estimates against reality. EXPLAIN ANALYZE shows both, and a plan whose estimate is wrong by orders of magnitude is a plan built on missing statistics — which is the real fix rather than a hint.

Key terms: The small-files problem, Compaction, Spill, Query execution plan, Flex execution class