One task runs for three hours while 199 finish in seconds
A PySpark notebook joins a 1.4 TB Delta fact to a 3 MB dim_product Delta table and writes a partitioned aggregate. The cluster is 8 workers x 8 cores / 64 GB, autoscaling 2-8, on DBR 15.4 LTS with AQE at defaults. The job takes 3 h 10 min. The Spark UI shows the shuffle stage with 200 tasks: 199 complete in under 40 seconds, one runs for 2 h 50 min and spills 38 GB to disk. The team's first fix was spark.sql.shuffle.partitions = 2000, which slowed the job to 4 h 05 min and produced 2,000 small output files. A reviewer also noticed the code calls spark.read.format('delta').load(dim_path).toPandas() and passes the resulting DataFrame into a Python UDF that maps category names.
1. Diagnose the single long task. Why did raising shuffle partitions make it worse, and what would you actually change?
Read the worked answer
One task holding almost all the time and spilling 38 GB is skew, not insufficient parallelism. Some join or group key is massively over-represented - typically a null or placeholder product_id, or a catch-all category - and hash partitioning sends every row with that key to one partition. Adding partitions cannot split a single hot key; it only adds scheduling overhead and, downstream, tiny files, which explains both the slower runtime and the 2,000 outputs.
Confirm it empirically: count rows per key on the fact, ORDER BY count DESC LIMIT 20, and compare the task's input size with the median. Then pick the remedy that matches the cause. If it is a join, confirm AQE skew handling is active (spark.sql.adaptive.skewJoin.enabled, skewedPartitionFactor, skewedPartitionThresholdInBytes) - it only helps sort-merge joins and only when the partition exceeds the byte threshold.
If the hot spot is an aggregation, AQE will not fix it: isolate the dominant keys, aggregate them separately and union, or salt the key with a random suffix and re-aggregate. And since the dimension is 3 MB, the right join should be a broadcast, which removes the shuffle entirely - see the toPandas problem.
2. What is wrong with reading the 3 MB dimension via `toPandas()` into a Python UDF, beyond style? What replaces it?
Read the worked answer
toPandas() collects the whole dimension to the driver, so the dimension never participates in the Spark plan as a table. Three consequences follow. The driver holds the data and every executor serializes rows to Python for the UDF, which costs far more than a native expression; the UDF is a black box to Catalyst and to Photon, so no predicate pushdown, no code generation, no vectorization; and because there is no join, the optimizer has no statistics to broadcast with, so the fact still shuffles.
The replacement is a real join. The dimension is 3 MB, comfortably under the default spark.sql.autoBroadcastJoinThreshold of 10 MB, so a plain join should broadcast automatically once table statistics exist - run ANALYZE TABLE or rely on Delta statistics collection, and add an explicit /*+ BROADCAST(dim_product) */ hint if the planner still guesses wrong. Map categories with a native when(...).otherwise(...) or a lookup join rather than a Python function.
If some transformation genuinely needs Python, use a pandas UDF or mapInPandas so work is vectorized in batches, and keep it off the join path.
3. The write produced 2,000 small files. How do you fix file sizing now, and what should the write pattern be going forward?
Read the worked answer
Two thousand files came from forcing 2,000 shuffle partitions and writing them straight out. Immediate remediation: run OPTIMIZE on the target table to compact small files to the target file size, and if the table is queried by a common predicate, add liquid clustering or ZORDER BY on that column during the same rewrite. Check DESCRIBE DETAIL for file counts before and after so the fix is measured, not assumed.
Going forward, stop setting spark.sql.shuffle.partitions manually. With AQE on, spark.sql.adaptive.coalescePartitions.enabled and spark.sql.adaptive.advisoryPartitionSizeInBytes (commonly 128 MB) size the post-shuffle partitions from actual data volume. If the final write must be partitioned, repartition explicitly by the partition column so one file set per partition is produced at a sane size, and never partition on a high-cardinality key like product_id.
Finally, enable Delta auto-compaction for the incremental writes and schedule OPTIMIZE on the table so small-file drift from frequent appends is corrected continuously rather than discovered in a cost review.