CMSC724: Parallel Execution and Skew
Instructor: Amol Deshpande
This chapter follows “Practical Skew Handling in Parallel Joins” (DeWitt, Naughton, Schneider, Seshadri; VLDB 1992).
Chapter 1 explained how operators cooperate within a plan. Here we ask how to divide an operator’s work across machines. The central challenge is skew: some workers receive much more work than others, so the slowest worker determines when the operation finishes. We first develop parallel join, sort, and aggregation, then examine ways to balance their work.
1. The Shared-Nothing Setting
In a shared-nothing system, each machine has its own processor, memory, and storage. A worker accesses local data directly and communicates with other workers to obtain remote data. This contrasts with cores sharing a machine’s memory; the same partitioning ideas can apply there, but communication costs differ.
A relation is split into partitions: disjoint subsets of its rows. Thus R = R₁ ∪ R₂ ∪ R₃, with each row initially held by one machine. A partition may be defined by a key range, a hash of a key, or simply where the data was loaded.
Having partitions is not enough for every operation. Rows that need to interact may reside on different machines. The execution plan must arrange for them to meet, while avoiding unnecessary communication. Throughout the join examples, assume an inner equality join and ordinary duplicate-preserving SQL semantics.
2. Parallel Hash Join
To join R and S on A, route rows of both relations using the same function, for example destination = h(A) mod p for p workers. This redistribution is a shuffle.
The correctness rule is simple: equal join keys must reach the same worker. Once redistribution finishes, worker i can join its local R and S partitions without consulting any other worker. It builds a local hash table and probes it as in Chapter 1. The routing hash and the local hash-table hash serve different purposes.
For example, with two workers and integer keys, routing by A mod 2 sends every key-7 row from both relations to worker 1. If there are two such rows in R and three in S, that worker emits all six matching pairs.
The combined result remains partitioned across workers. A following operator may use that distribution, request another shuffle, or gather the output.
Communication can dominate computation. A shuffle consumes network bandwidth, serialization work, buffering, and often disk I/O. Within one machine, repartitioning still costs memory traffic and synchronization. If both inputs already have compatible partitioning on the join key, the plan may avoid redistribution entirely.
The asymmetric variant
A broadcast join leaves the large relation R in place and sends a complete copy of the smaller relation S to each worker. Each worker joins its local R partition against all of S. Every result pair is produced by the worker that owns its R row.
This avoids shuffling the large input, but duplicates transmission and storage of S. It is attractive when S is small enough to fit in each worker’s memory and broadcasting it costs less than repartitioning both inputs. There is no universal size-ratio threshold.
Broadcasting avoids concentrating all rows for a frequent key through a key-based shuffle. It does not eliminate skew: initially uneven R partitions or different numbers of matches per row can still leave one worker with more work.
3. Parallel Sort
For a global sort, hash partitioning is insufficient: it groups equal keys but does not order partitions relative to one another. Instead, choose ordered range boundaries, route each row to its range, and sort locally.
For example, workers may own A < 100, 100 ≤ A < 200, and A ≥ 200. Concatenating their sorted outputs in range order gives a globally sorted sequence. The outputs need not be gathered into one file unless the consumer requires it.
Equal-width ranges need not contain equal numbers of rows. If most keys lie below 100, the first worker does nearly all the sorting. Sampling can estimate quantiles—boundaries intended to put roughly equal numbers of rows in each partition. Repeated values need special care if one value alone exceeds a worker’s target share.
4. Group-By and Aggregation
A group-by requires all contributions to a group’s result to be combined. Unlike a join, many aggregates can reduce the data before it moves across the network.
For SUM(amount) GROUP BY region, each worker computes a partial aggregate for every region in its local input. A local partial of 9 and a remote partial of 10 combine to 19. Millions of input rows may become only a few partial sums.
If there are few groups, gathering the partials to one worker can be inexpensive. If there are many, hash the partial states by group key and let several workers perform the final combination.
Partial aggregation reduces communication only when local groups contain several rows. If almost every row has a distinct key, little reduction is possible. A frequent key is often less troublesome here than in a join: its many rows can become one small partial state per worker.
Which aggregates decompose
The requirement is a mergeable state: each worker can summarize its rows, and those summaries can be combined correctly.
| Aggregate | Local state | Final combination |
|---|---|---|
SUM, COUNT |
A sum or count | Add the partials |
MIN, MAX |
A local extreme | Take the extreme of the partials |
AVG |
Sum and count | Add both, then divide |
Averaging averages is generally wrong: workers may contribute different numbers of rows. If one worker has (sum=10, count=1) and another (sum=180, count=9), the final average is 190/10 = 19. Averaging the local averages instead gives (10 + 20)/2 = 15.
Exact median does not admit a fixed-size summary like sum and count. In particular, medians of partitions do not determine the global median. Exact algorithms may use sorting or distributed selection rather than gathering the entire group at one machine. Approximate quantiles can use compact mergeable sketches. The distinction is whether a small sufficient state exists, not whether any parallel algorithm exists.
5. Four Kinds of Skew
A straggler is a worker that finishes much later than the others. For a stage that waits for every worker, completion time is approximately the maximum worker time, not the average. Even with three equal-speed machines, distributing work as 80, 10, 10 leaves two machines idle while the first finishes.
The paper distinguishes four sources of imbalance:
| Kind of skew | Where it arises | Example |
|---|---|---|
| Tuple placement | Initial storage | One machine initially holds most rows |
| Selectivity | After filtering | A predicate keeps 90% on one worker and 1% on another |
| Redistribution | During routing | A frequent join key sends many rows to one worker |
| Join product | While producing matches | Similar input sizes yield very different output sizes |
Selectivity here means the fraction of rows that satisfy a predicate. Join product skew requires looking beyond input sizes: if a key has r rows in one input and s in the other, it produces r × s result pairs. A key appearing 1,000 times in each input creates one million pairs. A good hash function spreads distinct keys; it cannot spread copies of the same key while preserving ordinary single-destination routing.
6. The Techniques
Sampled range partitioning
Choose range boundaries from a sample so partitions contain similar amounts of data, rather than similar widths of key space. Range routing and the local join algorithm are independent: range-partitioned inputs can still be joined with hash tables.
The paper emphasizes balancing the build relation because an oversized build partition can spill and incur extra I/O. Probe input size and output size still matter; build balance alone is not a complete model of work.
Subset-replicate: splitting a frequent key
No ordinary hash or range function can split the six 1s in {1,1,1,1,1,1,2,3} across workers. Subset-replicate relaxes the rule that a key has just one destination.
Split the build rows into disjoint subsets, such as {1,1,1,1} and {1,1,2,3}. Then replicate every key-1 probe row to both workers. Each build row lives at one worker, so every matching pair is emitted exactly once.
The correctness requirement is essential: splitting both sides independently would miss cross-worker pairs; replicating both sides in full would duplicate them. Splitting one side and replicating the other trades communication for balanced computation.
Weighting: choosing the subset sizes
Suppose the build relation is to be split into three four-row partitions:
worker 1: {1,2,3,4}
worker 2: {4,4,4,4}
worker 3: {4,4,4,6}The eight key-4 build rows should be assigned in proportions 1/8, 1/2, and 3/8, because the first and third workers already have other rows. These weights guide assignment of rows on the subsetted side. Matching probe rows still need replication to every destination containing key 4. Weighting refines load balance; it does not remove that correctness requirement. See the paper’s Section 2.2.
Virtual processors: more partitions than workers
Create many small logical partitions—say, 100p for p workers—and map several to each physical worker. These partitions are called virtual processors; they are units of work, not extra hardware.
The paper considers round-robin assignment and assignment using estimated join costs. Many small pieces reduce the chance that one expensive range determines an entire worker’s load. Frequent keys may still need subset-replicate so that their work can actually be divided.
The abstraction also permits more flexible scheduling, but distinguish that possibility from the paper’s concrete schemes: its round-robin mapping is predetermined. Over-partitioning alone cannot break up a single indivisible heavy task or guarantee balance without useful partition boundaries.
7. Sampling, in 1989
Range boundaries depend on samples, so obtaining representative samples is part of the algorithm. With disk-resident data, reading scattered individual rows can require many expensive random I/Os.
Stratified sampling divides the population into strata—here, the rows on each machine—and samples within each. It can be a sound sampling method. However, taking the same number of samples from machines with different row counts overrepresents the smaller partitions unless inclusion probabilities or weights are adjusted.
Page-level sampling reads a randomly chosen page and uses its rows, amortizing I/O over several samples. Nearby rows may have correlated keys, reducing the information gained compared with independently sampled rows. Page occupancy also matters when the target is a uniform sample of tuples.
The lesson is to distinguish a cheap sample from a representative one: a poor sample can choose poor boundaries before the join even starts.
8. Implementation in Gamma
The techniques were implemented in Gamma, an early parallel database prototype whose research was very influential on commercial systems.
Gamma uses a split table to drive repartitioning. The paper adds a new split table type for range partitioning, and notes that weighted range partitioning drops in easily.
A split table maps a hash bucket or key range to a destination worker. Gamma originally associated one outgoing buffer page with each entry. Making many more virtual partitions therefore made the buffer allocation, rather than just the routing metadata, too large. A two-level table separates logical ranges from physical destinations, allowing ranges going to the same worker to share an outgoing buffer.
9. Results and Recommendation
The findings:
- Basic hybrid hash works well under low skew, but severe skew can cause excessive overflow I/O and very poor completion times in the experiments.
- Virtual processor range partitioning with round robin works best across the widest range of scenarios.
Their final recommendation, which is admirably concrete:
- Take a sample to see which relation, if either, is skewed.
- If neither is skewed, use the basic parallel hash join — avoid the overhead of everything else.
- If one is skewed, use virtual processor range partitioning with round robin, with the more skewed relation as the build relation.
Note the shape of that recommendation: measure first, and use the cheap thing when you can. The sophisticated technique is the fallback, not the default.
10. Why This Is Still Current
The same challenge appears in newer systems. Spark’s adaptive query execution documentation describes splitting skewed join partitions and replicating matching partitions when necessary, a feature introduced in Spark 3.0 in 2020. This is the same correctness pattern as subset-replicate, though it does not establish a direct historical lineage.
Over-partitioning similarly separates logical pieces of work from physical workers. The useful connection to Chapter 4 is that smaller schedulable units let a system distribute work more flexibly. But task scheduling and key skew are distinct: scheduling cannot balance a huge hot-key join unless that join’s work is divisible.
When diagnosing a parallel operator, ask three questions: Are the input bytes balanced? Is the computation per input row balanced? Is the output balanced? Equal partition sizes answer only the first. Sampling, replication, and scheduling address different parts of the problem, so they are often used together.