9.2 Partitioning, Shuffle, and Data Skew: Understanding Distributed Batch Processing Plans

After distributing the equipment logs among 20 workers, filtering is fast, but aggregation using equipment_id stalls on the last few tasks. Most partitions finish in just a few tens of seconds, while the partition containing the "unknown" key runs for 40 minutes. Adding more machines doesn't shorten the long tail, because all records with the same key still need to be consolidated.

The core of distributed performance isn't the number of nodes; it's how data moves, aggregates, and recalculates in the face of node failures.

This lesson's objectives

• Understand the relationship between job, stage, task, and partition;
• Distinguish between narrow dependencies and wide dependencies that require shuffle;
• Choose a join strategy based on table size and distribution;
• Identify skew, spill, and small tasks from the execution plan and runtime metrics.

1. From Logical Query to Physical Execution

For example, with Spark SQL, a user writes a logical plan involving scan, filter, join, and aggregation. The optimizer selects physical operators and splits the execution graph into stages; each stage contains parallelizable tasks, with each task typically processing one partition.

from pyspark.sql import functions as F

missions = spark.read.parquet("missions/")
equipment = spark.read.parquet("equipment/")

result = (
    missions
    .filter(F.col("event_date") >= "2026-01-01")
    .join(equipment, "equipment_id", "left")
    .groupBy("equipment_type")
    .agg(
        F.count("*").alias("mission_count"),
        F.avg("wear_rate").alias("avg_wear_rate"),
    )
)

result.explain(mode="formatted")

First, check that the plan only reads necessary columns, the filter is close to a scan, and the join type and statistics estimates are reasonable, then tune the configuration.

2. Narrow Dependencies and Wide Dependencies

Operations like filtering and row-wise mapping typically depend only on the current partition and can be executed in a pipeline. However, operations such as key-based aggregation, deduplication, sorting, and joining large tables usually require that records with the same key end up in the same target partition, triggering a shuffle.

shuffle includes:

• Bucket and write the records again upstream;
• Transmit partition data over the network;
• Read, sort, or construct hash state downstream;
• Retry or recalculate upon failure.

It's not about banning shuffle operations, but about reducing meaningless data movement. Filtering, projecting, and local pre-aggregation first can lower shuffle bytes; if the final semantics require key-wise aggregation, shuffle itself can't simply vanish just because of a slogan.

3. The number of partitions is a trade-off between parallelism and overhead

Too few partitions:

• Insufficient parallelism;
• High memory pressure within a single task;
• Failures are recalculated in large units.

Too many partitions:

• High scheduling and serialization overhead;
• Generates a large number of small files;
• Increased metadata and object storage requests.

Don't set it based solely on input bytes. Monitor each task's input, shuffle, duration, and spill, and then consider worker cores, memory, compression ratio, and operator expansion. Modern engines' adaptive execution can dynamically merge small shuffle partitions or handle partial skew during runtime, but still rely on statistics and reasonable task layout.

repartition typically redistributes data and triggers a shuffle; coalesce is often used to reduce partitions, avoiding full reorganization, but may result in imbalance. The two are not synonymous with "changing file count."

4. Join First verify semantics, then optimize strategy

Performance optimization can't hide join fanout. First, verify key uniqueness, NULL semantics, expected row counts, and the proportion of unmatched rows.

Common physical strategies:

Broadcast hash join

Broadcast a small side to the worker, join locally with the large table to avoid key shuffling of the large table. Whether "small" is sufficient depends on serialized size, concurrency, worker memory, and engine thresholds; don't blindly force broadcast based on disk file size.

Shuffle hash/sort-merge join

Redistribute partitions on both sides using a join key, then connect the corresponding partitions. Suitable when both sides are large, but network and disk costs are high.

Use Existing Layout

If the tables' partitioning, bucketing, or sorting are compatible, the engine might reduce shuffle. It's important to verify that the execution plan actually leverages the layout, rather than assume it based on table definitions alone.

Collecting table and column statistics in a timely manner helps the optimizer estimate row counts and choose optimal strategies. Hints are a last resort and do not guarantee effectiveness for all join types.

5. Why Data Skew Creates Long Tails

Tilt may originate from:

• NULL, unknown, or the default value form a superkey;
• A small number of large customers/"fortress" naturally hold the majority of records;
• The filtered distribution differs from historical statistics;
• join fanout causes certain keys to produce massive Cartesian products;
• Partition hashing is uneven or input file sizes are significantly different.

Diagnosis must consider the task distribution, rather than stage averages alone: maximum/median task durations, input rows, shuffle bytes, spills, and output rows.

Optional fix:

• Separate and handle NULL/unknown individually if they are distinct in a business context;
• Perform a provable mergeable two-phase aggregation on hot keys;
• Use salting for tilted joins and ensure proper deduplication and de-salting;
• Broadcast the small table truly;
• Enable and verify the engine's adaptive tilt processing;
• Fix the upstream fanout, rather than adding machines for erroneous results.

6. Memory, Spill, and Cache

The execution engine can spill intermediate state to disk to avoid OOM, but spilling increases I/O and does not imply memory independence. You must simultaneously monitor execution memory, caching, GC, disk space, and serialization.

Caching is suitable for intermediate results that are expensive to recalculate and reused multiple times. Caching data used only once wastes resources; caching raw large tables might perform worse than relying on columnar scans. After caching, it's essential to verify a cache hit and release the cache when its lifecycle ends.

7. Retry Requirements for Deterministic and Idempotent Output

Distributed tasks may be retried due to worker failures or speculative execution. Pure transformations must produce the same output given the same input; however, actions within UDFs such as sending emails, making payments, or appending external records can lead to unintended side effects when retries occur.

A strategy should include:

• Assignment run ID and deterministic target partition;
• Write to a temporary location and atomically/transactionally publish;
• Retries overwrite the same batch rather than appending repeatedly;
• Recoverable states before and after submission;
• Input, code, configuration, and output version history.

"A task succeeding once" doesn't mean the entire table is published exactly once; the submission protocol and downstream storage semantics are equally important.

8. Performance Troubleshooting Order

1. Validate row counts, uniqueness, and business outcomes
2. View logical/physical plans and statistics estimates
3. Find the slowest stage and the longest tasks
4. Distinguish scan, shuffle, CPU, spill, GC, and write
5. Check the maximum rather than just the average
6. Change one variable at a time and repeat the representative baseline
7. Record costs, stability, and result validation

Common Misconceptions

-More nodes, faster: A single hot key or sequential stage won't scale linearly.

• Try to avoid all shuffle operations: Some global semantics inherently require data rearrangement.
• More partitions, more parallelism: Tiny tasks and small files eat up the benefits.
• Caching Doesn't Always Speed Things Up: Data that's read once or can be efficiently rescanned might actually run slower.

Practice

1. Identify possible stage boundaries for filtering, group by, global sorting, and join.
2. Construct an unknown hotkey to compare average versus maximum task metrics.
3. Compare the execution plans, network bytes, and peak memory usage of broadcast join and shuffle join.
4. Design a safe-retry submission protocol for partitions.

Summary

Partitions provide parallel units, shuffle is responsible for reestablishing key relationships, and skew pulls the parallel job back into the few long-tail keys. Only by understanding the plan and task distribution can you effectively optimize data movement, state, and publish semantics.

Next lesson: Handling unbounded data, stream processing must manage event time, late data, replay, and end-to-end delivery guarantees.