Flink in Practice: From Task Slots to Hot Keys

Category: Categories: Study Notes

In a multi-tenant B2B system, customers rarely generate the same amount of traffic. Many send a steady trickle of events; a few contribute a much larger share. A streaming pipeline has to handle both, often with the same set of workers.

In Flink, that imbalance can be easy to miss at first. The cluster may have CPU and memory to spare while a small number of tasks struggle to keep up. Adding workers seems like a reasonable next step, but whether it helps depends on how the work is divided—and which parts can be divided at all.

Before choosing more workers or larger ones, it helps to follow a record through the job. A Kafka-to-Kafka metrics pipeline gives us a concrete example: the path from an incoming event to an aggregate connects Flink’s architecture to the choices we make about keys, parallelism, and worker size.

Following a record through the job

Consider a record containing a tenant ID, a metric name, some dimensions, and a value. A Kafka source reads it, a parser turns it into an application object, an aggregation updates the appropriate metric, and a sink writes the result. These steps are the job’s operators.

The JobManager coordinates the job: it arranges execution, coordinates checkpoints, and handles recovery. The records themselves move between TaskManagers, the worker JVMs where the operators run. They don’t pass through the JobManager on their way from source to sink.

Flink's historical runtime architecture, showing the client, JobManager, and TaskManagers.

Figure 1. The JobManager coordinates execution; TaskManagers run the tasks and exchange records. This historical diagram includes Actor System labels from the 2016 implementation.

In today’s architecture, those coordination responsibilities are divided among the Dispatcher, ResourceManager, and a JobMaster for each job. For sizing a pipeline, the important separation is still the same: the JobManager coordinates work that executes elsewhere.

Suppose we give the aggregation a parallelism of 16. Flink creates 16 parallel instances, called subtasks, each responsible for part of the stream. A record goes to one of those instances; it isn’t processed by all 16.

Some adjacent steps can run together. If parsing can be chained with the source, Flink avoids a separate handoff between them. The chained operator instances execute as one task on a task thread. This is why counting the boxes in application code doesn’t tell us how many threads the job will run.

A WordCount pipeline expanded into parallel subtasks, with aggregation and sink chained together.

Figure 2. An operator graph expands into parallel subtasks. Compatible adjacent steps can be chained into one task.

Those tasks need somewhere to run. A TaskManager offers slots for scheduling them, and a slot can hold tasks from several stages of the same job through slot sharing. Sharing a slot doesn’t chain those tasks together, and it doesn’t give them a dedicated CPU core.

Six slots across two TaskManagers, with tasks from different stages sharing slots.

Figure 3. Tasks from different stages share six slots across two TaskManagers. The particular placement is illustrative.

For example, a source at parallelism 12, an aggregation at 16, and a sink at 8 can fit into 16 slots under the usual single slot-sharing group, provided their resource requirements are compatible. Separate groups can need more. At two slots per TaskManager, that gives us a minimum of eight workers to schedule this job. Whether those workers can keep up is a separate calculation.

The key decides where the work ends up

Sixteen aggregation subtasks are useful only if the records give them enough independent work. For the metrics example, this comes down to the application key: the business key passed to keyBy, or expressed in a SQL GROUP BY.

Grouping by tenant_id is an obvious choice. It also means that all updates for a tenant converge on one aggregation subtask. A larger customer can therefore overload its owner while other subtasks have much less to do.

If the product actually computes a separate result for each metric series, the grouping can be more specific:

1	(tenant_id, metric_name, normalized_dimensions)

Now different series from the same customer can be handled independently. The dimensions still have to describe the metric we want to compute. Adding an arbitrary field just to spread the load changes the result unless another stage merges it back together. Dimension ordering also needs to be consistent, and unbounded dimension values can make state grow quickly.

Flink organizes keyed state into key groups and assigns those groups to subtasks. Rescaling moves groups between owners; it doesn’t divide one key among several owners at the same operator. The key-group count is set by maximum parallelism, so that setting deserves some thought before a stateful job needs to grow.

This is the distinction to keep in mind when choosing parallelism. Many moderately busy keys can spread out. One extremely busy key still arrives at one place.

Kafka adds two more places to look

The grouping inside Flink is only one part of the route. The input topic has its own partitions, and the Kafka key chosen by the producer doesn’t automatically become the application’s Flink key.

With 12 input partitions, at most 12 source readers have partitions to read concurrently. Raising source parallelism to 24 won’t split those partitions. It can still make sense to run a downstream stage at a different parallelism, though, because that stage may do much more work per record. A balanced input topic can also become badly skewed after keyBy.

At the other end, with key-based Kafka partitioning, the output key determines where writes go. Adding sink subtasks won’t remove a bottleneck caused by concentrating writes on one Kafka partition. With event-time windows, idle source readers need attention too: without idle detection, they can hold back watermarks and delay results even when the active readers are keeping up.

Following a record through Kafka and Flink also means accounting for what happens during recovery. A completed checkpoint records source positions together with the corresponding operator state. After a failure, Flink restores that state and resumes from those positions. Kafka’s committed consumer offsets can tell us about progress, but they aren’t the authority for restoring the job. A fresh start without restored state needs an explicit starting-offset policy.

Kafka output requires its own configuration. For exactly-once delivery, checkpointing must be enabled, the sink must use transactions, concurrently running applications need distinct transaction prefixes, and consumers must read committed data. Transaction timeouts need to cover checkpoint and recovery delays. Since committed output becomes visible with checkpoint completion, that delay belongs in the latency budget too.

Turning the workload into a cluster size

Once we know how records reach each stage, the sizing question becomes more concrete. We need a measured rate that a subtask can sustain with the real record format, state backend, checkpointing, and sink behavior. A benchmark that skips state updates or writes to a dummy sink can give a very misleading answer.

For a stage with reasonably balanced input, a first estimate is:

1 2	parallelism ≥ ceil(peak input rate / (capacity per subtask × target utilization))

For example, a stage receiving 48K events per second would need 16 subtasks if each had a measured capacity of 4K events per second and we targeted 75% utilization. This is an illustrative calculation. The measured capacity needs to reflect the resources each subtask will actually get when sharing a worker.

Throughput alone still leaves a lot out. Record size affects network traffic. The number of active keys and the state kept for each one affect storage. Frequent state access can make serialization or disk bandwidth matter long before memory is full. CPU and memory are part of that picture, alongside network and disk capacity.

There also needs to be room to catch up. If live traffic arrives at rate R and a restart leaves B records behind, clearing that backlog in T seconds requires more than R + B/T processing capacity, with restore time accounted for separately. A cluster that only just handles normal traffic will struggle to recover from an interruption.

For two-vCPU workers, two slots per worker are a reasonable place to begin measuring a CPU-heavy job. Slot sharing may put several runnable task threads on that worker, alongside connector and state-backend threads. Increasing the slot count creates more scheduling capacity without adding CPU.

Memory needs similar care. An illustrative 16 GiB process budget could be configured as:

1 2	taskmanager.numberOfTaskSlots: 2 taskmanager.memory.process.size: 16g

CPU allocation is configured in the deployment environment. The 16 GiB covers the whole process: heap, managed memory, network buffers, other native memory, and JVM overhead. It doesn’t become a 16 GiB heap or two isolated 8 GiB heaps. How that budget is divided needs to match the state backend and fit within the actual container limit.

The JobManager has a different workload. Its resources support coordination, the execution graph, and checkpoint metadata. Giving it more CPU won’t speed up an aggregation running on a TaskManager.

When a few tasks become the bottleneck

The sizing estimate assumes that work spreads reasonably evenly. During testing of a Kafka-to-Kafka metrics job, we observed a few tasks reaching 100% busy time while their TaskManagers still had CPU and memory to spare. Looking only at worker utilization would have missed how little capacity remained in those tasks.

The execution model makes this less surprising. One CPU-bound task thread can occupy a full core on a two-core worker while total CPU utilization is around 50%. That is one way spare CPU and a saturated task can coexist. A thread profile is still needed to establish what keeps the task occupied.

Busy time needs a little interpretation as well. A busyTimeMsPerSecond value near 1,000 means the task is spending almost none of its measured time idle or waiting for Flink output buffers. It isn’t a CPU percentage. Synchronous work that waits on a dependency can occupy a task without consuming much CPU, and a busy downstream task can cause backpressure further upstream.

That gives us a more useful investigation than watching the worker averages. Follow the slow path back from the sink, compare individual subtask rates, and inspect the busy task’s thread profile. State access, serialization, timers, and sink waits are all possible places to spend time. The diagnosis needs to explain both where records accumulate and what keeps the receiving task occupied.

Combining updates before the hot key

If the bottleneck is an aggregation receiving too many updates for the same key, there is another option besides adding resources: reduce the number of updates that reach it. An upstream aggregator can combine raw metric updates and send partial results to the final owner.

For a sum, many updates to the same metric can become one partial sum. The final owner still owns the key, but it receives fewer updates to process. Flink’s local–global aggregation follows this pattern: combine repeated work locally before the keyed shuffle, then merge globally. The gain depends on how often keys repeat within a batch. Mostly unique keys leave much less to combine.

Before: raw red-key records converge on a hot aggregation task. After: parallel local aggregators produce partial sums 17, 18, and 15, which are merged by a global aggregator.

Figure 4. The red key receives twelve raw updates on the left. On the right, three local aggregators turn them into three partial sums before the shuffle. The global aggregator still owns the key, but it has fewer records to merge.

For eligible Table/SQL aggregates, the official tuning example enables mini-batching and two-phase aggregation like this:

SET 'table.exec.mini-batch.enabled' = 'true';
SET 'table.exec.mini-batch.allow-latency' = '5 s';
SET 'table.exec.mini-batch.size' = '5000';
SET 'table.optimizer.agg-phase-strategy' = 'TWO_PHASE';

The five-second setting is an example to tune against a latency budget, not a value to copy into every pipeline. Check EXPLAIN to see whether local and global stages are actually present; unsupported aggregates can fall back to one phase. A custom DataStream job needs the equivalent stages implemented explicitly.

The merge has to preserve the metric’s meaning. For an average, send (sum, count) and divide after merging; averaging partial averages loses their weights. Distinct counts and percentiles need suitable mergeable representations. For windowed results, carry the window identity and decide how late updates are applied. Adding a cumulative partial total twice will count the same events twice.

An upstream buffer also becomes part of the job’s state. It must be bounded and included in checkpoints, with restoration and rescaling tested. Buffered events must respect watermark and window semantics. Moving the aggregation earlier should still produce the same answers after a failure.

Placement matters just as much. Putting the extra aggregator after the existing hot grouping would leave the original bottleneck in place. If input distribution is already concentrated on too few workers, repartitioning or a two-stage salted key may be needed first. The salt must vary within the hot key, and the final stage must merge it away.

The next test should compare the raw input rate with the rate of partial results reaching the global stage, then check the busiest subtasks, lag, latency, checkpoints, and output correctness during a replay. That tells us whether pre-aggregation has removed enough repeated work to be useful. The global stage still has one owner per key; it simply has fewer updates to handle.

References and image credits

The architecture and resource discussion draws on these four posts from WuChong (Jark)’s blog:

WuChong (Jark). Flink 原理与实现：架构和拓扑概览. May 3, 2016. Source of Figure 1.
WuChong (Jark). Flink 原理与实现：理解 Flink 中的计算资源. May 9, 2016. Source of Figures 2–3.
Fabian Hueske; Chinese translation by WuChong (Jark). 确定 Flink 作业所需资源大小时要考虑的 6 件事. October 30, 2018.
Fabian Hueske and Markos Sfikas; Chinese translation by WuChong (Jark). Flink 如何管理 Kafka 消费位点. November 4, 2018.