A stream processor looks like a batch engine from a distance. Both take a program, translate it into a graph, and run the graph across a cluster. The differences only appear once you accept that the job never ends, and almost every architectural decision in a stream processor follows from that one fact.

Three processes, three jobs

A running cluster has a coordinator, some workers, and a client that submitted the job and may well have gone away.

The coordinator is usually called the job manager. It holds the execution graph, decides which task runs in which slot, triggers checkpoints, and decides what to do when something fails. It is the only component that knows the global state of the job, and it is therefore the component whose loss is most serious. In a production deployment it runs with a standby and a leader election, because a coordinator that dies without one takes the job with it.

The workers are called task managers. Each one is a process with a fixed amount of memory, a fixed number of slots, and a network stack for exchanging records with its peers. A task manager runs the actual user code, holds the actual state, and writes the actual checkpoints. Everything that costs money is here.

The client translates your program into a graph and submits it. In the older session mode the client stayed attached and mattered; in the application mode that almost everybody now runs, the client's work finishes the moment the graph is submitted and the job no longer depends on it. This distinction matters when you are deciding what your deployment pipeline needs to keep alive.

The slot, and why it is not a core

A slot is the unit of resource isolation inside a task manager, and it is the single most misunderstood concept in this architecture.

A slot is a share of the task manager's managed memory. It is not a share of its processor. A task manager with four slots divides its managed memory four ways, and all four slots compete freely for all the available cores. This means slot count is a memory decision, and the common mistake of setting slots equal to cores produces a cluster that either wastes memory or runs out of it, depending on which way the job is shaped.

A slot can hold more than one task. In fact in a normal job it holds one task from every stage of the pipeline, which is what slot sharing means and which we will come to shortly. The practical consequence is that the number of slots you need is determined by the parallelism of the widest operator, not by the number of operators in the job.

Parallelism, subtasks and tasks

The vocabulary here is worth getting exactly right, because the metrics you will read later are named after it.

An operator is a logical step in your program, such as a map, a keyed aggregation or a sink. Its parallelism is the number of copies of it that will run. Each copy is a subtask. A task is the runtime unit that actually gets scheduled into a slot, and it may contain several chained operators rather than just one.

So a program with five operators at a parallelism of thirty two produces a hundred and sixty subtasks, which the planner may fuse down to perhaps three tasks per parallel instance, giving ninety six scheduled tasks occupying thirty two slots. Reading that sentence and knowing which number is which is most of what you need to interpret a web interface that is showing you an unfamiliar job.

Parallelism can be set globally, per operator, or left to a default. Setting it per operator is how you stop a sink that can only tolerate eight concurrent connections from being created thirty two times, and it is also how you accidentally create a bottleneck that the global number hides.

Operator chaining, and the cost it removes

When two consecutive operators have the same parallelism and the connection between them is a simple forward, the planner fuses them into one task. The record then moves between them as a method call on the same thread rather than through a serialisation boundary.

The saving is large and it is easy to underestimate. An unchained hop serialises the record, copies it into a network buffer, possibly sends it across the network, deserialises it and hands it to a different thread. A chained hop is a function call. In a pipeline doing simple per record work, chaining is frequently the difference between a job that keeps up and one that does not.

Chaining breaks whenever the parallelism changes, whenever the connection is a keyed exchange or a rebalance, or whenever you disable it deliberately. Disabling it deliberately is sometimes correct, because a chain is also a unit of blame: every operator in one chain reports as one task, so a single expensive operator inside a long chain is invisible in the metrics. Splitting a chain to find out where time goes is a legitimate debugging move, as long as you put it back.

Slot sharing, and the shape it produces

By default all tasks belonging to one job can share a slot, provided they come from different stages of the pipeline.

This produces an arrangement that surprises people the first time they see it. Each slot ends up holding a vertical slice of the entire pipeline: one source subtask, one map subtask, one window subtask, one sink subtask. The number of slots required equals the maximum parallelism of any single operator, not the sum of all the parallelisms.

The arrangement is also what makes a pipeline resilient to uneven work. A source subtask that is idle while a window subtask is busy are sharing the same processor allowance, so the idle one costs nothing. If every stage had its own dedicated slots, the idle stages would be holding memory and cores that the busy stage needed.

You can opt out with slot sharing groups, and the one case where you should is when a specific operator has a genuinely different memory profile from everything else. A large keyed state operator sharing a slot with four cheap stateless operators is fine. Two large keyed state operators sharing one slot will exhaust its memory, and separating them into their own groups is the fix.

What a running job actually looks like

It is worth making this concrete with numbers you can hold on to.

Take a cluster of twelve task managers, each with four slots, for a total of forty eight slots. Run one job at a parallelism of thirty two. Thirty two slots are occupied, sixteen are free, and each occupied slot holds a vertical slice of the pipeline. The job's five operators have fused into three tasks, so ninety six tasks are scheduled.

Now the important part. Those thirty two slots are spread across twelve machines, which means a keyed exchange between two stages is mostly network traffic between machines. The sixteen free slots are not spare capacity for this job, because its parallelism is fixed at thirty two and will not grow to fill them. They are capacity for a second job, or headroom for a rescale, and if they are neither of those things then they are waste you are paying for.

Failure, and who notices first

When a task manager dies, every task on it dies with it. Because the tasks in a pipeline are connected, the job manager cancels the whole job and restarts it from the last checkpoint. This is the part newcomers find startling: the failure of one worker out of twelve restarts all of them.

Fine grained recovery exists for jobs whose graph has independent regions, but a typical pipeline with a keyed exchange is one connected region, so in practice a single lost worker means a full restart. The restart is cheap in processor terms and expensive in state terms, which is why the next sections are about state and checkpoints rather than about scheduling.

When the job manager dies without a standby, the job stops and does not restart itself. With a standby and a highly available metadata store, a new leader picks up the execution graph and resumes from the last checkpoint. The difference between those two sentences is the entire argument for running the coordinator in a highly available configuration, and it costs one extra process.