Source-linked AI summary
The Power of Both Choices: Practical Load Balancing for Distributed Stream Processing Engines
Muhammad Anis Uddin Nasir, Gianmarco De Francisci Morales, David García-Soriano, Nicolas Kourtellis, Marco Serafini
TL;DR
Distributed stream processing engines need scalable load balancing for skewed stateful workloads. The paper introduces Partial Key Grouping, which combines key splitting and local load estimation to adapt the power of two choices. On Apache Storm, it improves throughput by up to 60% and latency by up to 45% compared with key grouping.
Problem
Skewed workloads overload some processing instances, while existing key-grouping and shuffle-grouping approaches impose load-balancing or state-management trade-offs.
Method
Partial Key Grouping combines key splitting and local load estimation to apply the power of two choices in distributed stream processing.
Results
On Apache Storm, Partial Key Grouping improves throughput by up to 60% and latency by up to 45% compared with key grouping.
Takeaways & Limitations
Partial Key Grouping provides better load balancing than key grouping while requiring less memory overhead than shuffle grouping.
Takeaways & Limitations
Partial Key Grouping is a programming primitive that cannot express every algorithm.
Abstract
from arXiv · showhide
We study the problem of load balancing in distributed stream processing engines, which is exacerbated in the presence of skew. We introduce Partial Key Grouping (PKG), a new stream partitioning scheme that adapts the classical "power of two choices" to a distributed streaming setting by leveraging two novel techniques: key splitting and local load estimation. In so doing, it achieves better load balancing than key grouping while being more scalable than shuffle grouping. We test PKG on several large datasets, both real-world and synthetic. Compared to standard hashing, PKG reduces the load imbalance by up to several orders of magnitude, and often achieves nearly-perfect load balance. This result translates into an improvement of up to 60% in throughput and up to 45% in latency when deployed on a real Storm cluster.
I. INTRODUCTION
Distributed stream processing engines must balance skewed stateful workloads online and with minimal coordination. The paper introduces Partial Key Grouping, combining key splitting with local load estimation to adapt the power of two choices while preserving scalability.
- Key grouping preserves per-key processing but can produce uneven message loads when key popularity is skewed.
- Skewed workloads can overload a few processing instances, causing poor resource utilization and inefficiency.
- A practical solution must operate online and distribute decisions with minimal coordination because key distributions are unknown in advance.
- Applying the power of two choices directly would require coordinating and storing a routing decision for every key, which is impractical for streams with billions of keys.
- Key splitting lets each key use both candidate processing instances, avoiding per-key agreement while retaining the power-of-two-choices approach.
- Partial Key Grouping combines key splitting and local load estimation into a stream partitioning scheme applicable to distributed stream processing engines.Local load estimates perform almost indistinguishably from a global load oracle.
- On Apache Storm, Partial Key Grouping improves throughput by up to 60% and latency by up to 45% compared with key grouping.
A. Existing Stream Partitioning Functions
Existing stream partitioning schemes trade off load balance against stateful processing requirements and scalability. Key grouping preserves per-key state but suffers under skew, while shuffle grouping balances messages at the cost of state aggregation and memory overhead; PoTC offers a promising but nontrivial alternative.
- Key grouping keeps messages with the same key on one PEI, typically using hashing.
- Shuffle grouping routes messages independently, usually round-robin, producing nearly equal message counts across PEIs but no per-key placement guarantee.
- Under skewed keys, key grouping can overload some PEIs and degrade service levels or cluster utilization.
- Stateful applications using shuffle grouping require multiple partial results per key, increasing memory and aggregation costs with parallelism.For the word-count example, memory and aggregation work can grow as O(WK), and approximation error may grow with the number of aggregations.
- Operator migration can rebalance load, but it requires application-specific parameters, state movement, and often substantial DSPE modifications.Coarse-grained partitioning in Storm and Samza does not directly support migrating individual keys between sub-streams.
- The power of two choices assigns each key to the less-loaded of two candidate workers, but requires load knowledge and tracking choices to preserve key grouping.Its theoretical load-balance gain over one choice is exponential, while more than two choices provide only constant-factor improvements.
A. Key Splitting
Key splitting relaxes key grouping by allowing each key to use two candidate PEIs, enabling decentralized power-of-two routing without per-key choice tracking. Combined with local load estimation, it supports dynamic balancing while limiting the memory and aggregation costs associated with shuffle grouping.
- Key splitting maps each key to both candidate workers and routes each message to the currently less-loaded candidate.
- This design lets multiple sources route messages independently in parallel without maintaining routing state.
- Dynamic load-based decisions let key splitting handle skew and adapt when key popularity changes over time.
- Compared with shuffle grouping, assigning each key to exactly two PEIs keeps state memory to a constant-factor increase and requires only one aggregation of two partial states.For streaming top-k word count, PKG uses O(K) memory and O(K) aggregation work, compared with O(WK) for shuffle grouping.
- Partial Key Grouping is the resulting partitioning scheme between key grouping and shuffle grouping, but it applies only to associative functions and monoids.Median computation is given as an example that cannot be easily expressed via PKG.
- Local load estimation avoids communication between sources and workers by having each source independently maintain worker-load estimates from its own stream portion.The overall maximum imbalance is bounded by the sum of the local imbalances.
IV. ANALYSIS
The analysis models partial key grouping as a colored balls-and-bins process and identifies when two-choice placement can or cannot maintain low imbalance. It then states bounds for Greedy-d under bounded maximum key probability.
- Model: PKG is analyzed with n workers, m sequential messages, irrevocable placements, and at most d workers processing each key.Keys are colors, messages are colored balls, and workers are bins; Greedy-d places each ball in the least-loaded hashed choice.
- Model: When d=1, Greedy-d models hash-based key grouping; when d is sufficiently large, it approaches shuffle grouping.The two endpoints connect single-choice key assignment with unrestricted choices across bins.
- Limits: If the maximum key probability p1 exceeds 2/n, expected imbalance grows linearly with m regardless of the placement scheme.The two candidate bins for the most frequent key receive at least a p1 fraction of messages in expectation, while average load grows at rate 1/n.
- Limits: Even p1 ≤ 2/n does not suffice: under a uniform distribution over n keys, two choices leave about 0.135n bins unused and yield imbalance of approximately 0.156m.The issue is an overpopulated set of potential-choice bins whose average load grows faster than the global average.
- Limits: Sampling variability can itself cause imbalance: with two equally likely keys, Greedy-2 has imbalance Ω(√m) with constant probability.Each key alternates across its two assigned bins, but the observed number of occurrences of each key fluctuates around its expectation.
- Results: For m ≥ n^2 and p1 ≤ 1/(5n), Theorem 4.1 gives a high-probability imbalance bound for Greedy-d when d ≥ 2.The analysis presents this as a significant reduction in maximum load and imbalance compared with key grouping; Theorem 4.2 states the bounds are best possible.
B. Proof
The proof develops concentration and subset-load arguments to bound the maximum load of Greedy-d. It separates lightly loaded bins from potentially overpopulated subsets and applies high-probability bounds to both.
- Concentration inequalities: The proof recalls Chernoff and McDiarmid concentration inequalities as tools for establishing the main theorem.These inequalities support bounds on sums of independent variables and on the number of heavily weighted bins.
- The μr measure of bin subsets: The measure μr(B) captures the probability that a key’s r hashed choices all lie within a bin subset B; μ1(B) measures one-choice containment.The proof uses μ1 and μd to relate key distributions to loads on bin subsets.
- The μr measure of bin subsets: For every bin subset B, the expected one-choice measure satisfies E[μ1(B)] = |B|/n.This expectation supports controlling the size and load contribution of subsets with unusually large one-choice mass.
- Scope of the bound: The stated bounds can be much smaller than worst-case estimates when most keys have probability substantially below p1.This observation qualifies the worst-case analysis for common distributions.
- Proof strategy: A coupling compares Greedy-d with a process that chooses bins uniformly at random, showing that the former’s relevant maximum-load events are no more likely.The comparison yields an O(m/n) bound with high probability for the maximum load in the analyzed subset.
- Proof strategy: Bins outside the heavy subset have maximum load at most 20m/n with high probability, while concentration bounds the heavy subset’s size.The proof then applies the subset-load lemma to obtain an O(m/n) bound there as well.
V. EVALUATION
The evaluation uses simulations and a real deployment to examine key splitting, local estimation, robustness, and PKG’s effect on applications in a real distributed stream-processing engine.
- Evaluation goals: The evaluation combines simulations with a real deployment of the proposed approach.The experiments are organized around four questions concerning key splitting, local estimation, robustness, and application-level effects.
- Evaluation goals: The experiments ask how key splitting affects PoTC and how local estimation compares with a global oracle.They also assess PKG’s robustness and its overall effect on applications deployed on a real DSPE.
A. Experimental Setup
The experimental setup uses large real, synthetic, and graph datasets chosen to represent varied skew and application domains. The datasets support tests of distributional shifts and robustness.
- Real datasets: The real datasets include Wikipedia page visits and Twitter words, selected for large size, differing skewness, and representative web and social-network workloads.Wikipedia keys are page URLs; Twitter keys are words extracted from tweets.
- Dataset summary: Table I summarizes each dataset by message count, key count, and the percentage of messages belonging to the most frequent key p1.These fields characterize workload size and key-frequency skew.
- Real datasets: A second Twitter dataset uses cashtags, whose popularity changes over time, to study shifts in key-distribution skew.Cashtags are ticker symbols preceded by a dollar sign, such as $AAPL.
- Synthetic datasets: Two synthetic datasets use log-normal key distributions with parameters derived from an Orkut analysis to emulate skewed online-social-network workloads.The datasets are named LN1 and LN2.
- Graph datasets: Three directed-graph datasets use edges as messages and vertices as keys to test robustness and represent streaming graph-mining applications.The datasets are LJ, SL1, and SL2.
B. Experimental Results
The experiments evaluate PKG against hashing, power-of-two-choices variants, and greedy methods, then test local load estimation, skew robustness, and deployment performance. PKG generally achieves very low imbalance and improves throughput and latency over key grouping, while avoiding shuffle grouping’s larger memory cost.
- PKG performs better than Off-Greedy in the reported simulations, achieving load balance comparable to offline algorithms after relaxing key grouping.
- PoTC alone does not guarantee good load balance; key splitting is fundamental for practical and effective streaming operation.
- Around 50 workers for WP and 100 for TW, imbalance transitions sharply when the worker count surpasses O(1/p1).
- Local estimation keeps imbalance within less than one order of magnitude of global information and remains robust to changes in source count.
- PKG remains robust to skewed source splits and drifting key popularity, with imbalance generally staying at very low absolute values despite occasional spikes.
- Compared with KG, PKG improves deployed throughput and latency while using about half the memory of SG.KG throughput falls by ≈60% under a tenfold CPU-delay increase, while PKG and SG fall by ≈37%; KG latency is up to 45% larger than PKG.
VI. APPLICATIONS
PKG is a stream-partitioning primitive that improves load balancing for expressible applications, but its applicability is limited by the algebraic structure required by the application.
- PKG is a novel stream-partitioning primitive, but not every algorithm can be expressed with it.
A. Na¨ıve Bayes Classifier
The naïve Bayes applications compare vertical key-grouped parallelism with horizontal shuffle-based parallelism and show how PKG reduces query-time coordination.
- Vertical parallelism tracks each feature independently but suffers from skewed feature distributions, causing load imbalance that PKG addresses.
- Horizontal parallelism distributes each feature-class count across workers, requiring query-time broadcasting because any worker may track a feature.
- PKG tracks each feature on two deterministically assigned workers, so queries probe two workers instead of broadcasting to all workers.
- The PKG query phase is less expensive and less sensitive to stragglers than the shuffle-grouping alternative.
- Streaming decision-tree implementations using shuffle grouping require memory that grows with W and expensive aggregation of W × D × C histograms.
- PKG uses 2×D×C×L histograms and merges only two histograms per feature-class-leaf triplet, making memory pressure reducible by adding workers.
C. Heavy Hitters and Space Saving
Heavy-hitter summaries support parallel stream processing, but merging introduces an additional error term. Key grouping avoids that parallelism-dependent error at the expense of poor load balancing, while PKG balances load with only two error terms.
- Heavy Hitters and Space Saving: SPACESAVING finds approximate top-k frequent stream items in constant time and space, with guarantees extendable to merged summaries.Merged summaries enable parallelized execution by combining partial summaries built independently on separate sub-streams.
- Heavy Hitters and Space Saving: The frequency-estimation error bound combines merging error with the sum of individual sub-stream summary errors for item i.Here, f_i is the true frequency, f̂_i the estimate, Δ_j each sub-stream error, and Δ_f the whole-stream merging error.
- Heavy Hitters and Space Saving: Key grouping makes an item's error equivalent to the sequential case because it depends on one summary, but it provides poor load balancing.The alternative bound depends on the parallelism level W when summaries are merged.
- Heavy Hitters and Space Saving: PKG balances workers while making each item's error depend on only two error terms, regardless of parallelism level, although individual bounds may depend on W.This combines load balancing with a smaller error dependence than general merged summaries.