Section 7: Where Each Tuple Goes, and What Happens When It Can't - Deep Dive: Stream Groupings, Parallelism & Backpressure
Executive Summary: A stream grouping is the routing function on each edge of the topology. Ours uses shuffle into the stateless parse bolts, fields by hashtag into the count bolts, and global into the single final ranker (Decision 2). Fields grouping makes the counts correct, and it's also the weak point. During the final, #worldcup reaches 150,000 tuples/s, all of it pinned to count executor 9, which handles 100,000. Partial aggregation fixes that. Then come parallelism (tasks are fixed at submit, executors aren't) and backpressure: how max.spout.pending and automatic throttling stop a fast spout from drowning a slow bolt (Decision 5).
Interviewer: "You've drawn arrows between the boxes. What does an arrow actually do?"
You: "Each arrow is a grouping, a function from a tuple to the task or tasks that receive it. It's where the developer tells the framework what the data means, so I pick each one on purpose."
Six Ways to Route a Tuple
The 2014 paper lists these groupings. Each answers one question per tuple: which downstream task gets it?
| Grouping | Routing | Use in our topology | Risk |
|---|---|---|---|
| Shuffle | Random, evenly balanced across tasks | Spout → parse (stateless) | Wrong for any bolt holding per-key state |
| Fields | hash(chosen fields) mod number of tasks | Parse → count and count → partial ranker, both by tag |
Hot keys; the mapping changes if the task count changes |
| All | Every task gets a copy | Broadcasting a config change, such as a new blocklist | Multiplies traffic by the task count |
| Global | The whole stream to one task | 16 partial rankers → 1 global ranker | Single-thread bottleneck unless the stream is tiny |
| Direct | The emitter names the target task | Custom, task-aware routing | The emitter must know the task layout |
| Local-or-shuffle | Same-worker tasks first; network only if there are none | Cheap hops where local capacity matches | Overloads a local task when the local source is faster |
Spout task 4 reads post p-77120 from partition 19 and shuffles it to a random parse executor. The parse bolt emits #worldcup and #goal, and fields grouping sends #worldcup to count executor 9 and #goal to count executor 2. Every #worldcup in the cluster lands on executor 9. So one plain in-memory hash map gives an exact count, with no locks and no distributed map (the alternatives Decision 2 rejected).
The count → partial ranker edge is fields-grouped by tag again, but count executors don't emit per increment. Every 5-s tick each count task emits its local top 1,000 candidates as (tag, windowTotal) tuples: 16 × 1,000 / 5 s = 3,200 tuples/s. Fields grouping on tag lands every tag in exactly one partial ranker, so no tag appears in two partial lists, whatever the layout. (With 16 tasks on both sides today, partial ranker 9 happens to read count task 9's tags, but correctness doesn't depend on that alignment.) Each partial ranker keeps its top 100, and the global ranker gets 16 lists of 100 every 5 s, a trivial load.
Interviewer: "Why not local-or-shuffle from spout to parse? It saves the network hop."
You: "A spout executor emits 37,500 posts/s and a parse executor handles 20,000. With about two executors per worker, a spout's worker holds at most one parse executor, so local-or-shuffle would push 37,500 posts/s into a 20,000/s thread. Locality pays only when the local consumers can absorb what the local producers emit. Shuffle costs a network hop and keeps all 40 parse executors evenly loaded."
Quick check before moving on:
Sign in to continue reading
The rest of this lesson is available with a free account. Signing in with Google or Microsoft is free.
Sign in to read the full lesson