Streams, Keys, and Operators

A record has to find its operator

Module 1 left GatewayMeter reading gateway-events, parsing each record into an Event, and counting events per model. This module adds the first computation that needs two events to meet: pairing a request's RequestStarted with its RequestCompleted into one RequestRecord. The job runs with parallelism 4, so there are four copies of every operator. Which copy sees which event, and who decides?

The answer is a routing decision you write into the program, and the same decision fixes where state may live. Get it wrong and nothing crashes: the job runs, the Web UI is green, and no request is ever assembled — which, for a metering job, means no request is ever billed.

Streams are partitioned, operators are parallel

Upstream's words, kept exact: during execution, a stream has one or more stream partitions, and each operator has one or more operator subtasks. The operator subtasks are independent of one another, and execute in different threads and possibly on different machines or containers. The number of operator subtasks is the parallelism of that particular operator.

Between two operators, records travel in one of two patterns. In upstream's words: One-to-one streams … preserve the partitioning and ordering of the elements — subtask 1 downstream sees what subtask 1 upstream produced, in that order. Redistributing streams … change the partitioning of streams. Each operator subtask sends data to different target subtasks, depending on the selected transformation. In a redistributing exchange, ordering is only preserved within each pair of sending and receiving subtasks.

The redistributing patterns are methods on a stream:

Only one row routes by the record's content: key_by. Everything else spreads load. The rest of the page is about what key_by does physically and why nothing else can substitute for it.

What key_by does physically

key_by(request_id) computes a key from every record and hashes it. The hash does not pick a subtask directly. It picks a key group, and the key group belongs to a subtask. Upstream's definition, kept exact: Keyed State is further organized into so-called Key Groups. Key Groups are the atomic unit by which Flink can redistribute Keyed State; there are exactly as many Key Groups as the defined maximum parallelism. During execution each parallel instance of a keyed operator works with the keys for one or more Key Groups.

Two numbers, then, and the spine's glossary insists you keep them apart. Parallelism is how many subtasks run now — 4 until module 10. Maximum parallelism is how many key groups exist, fixed for the life of the job's state. Unset, upstream gives the default as roughly operatorParallelism + (operatorParallelism / 2) with a lower bound of 128 and an upper bound of 32768 (as of 2026-09; see the upstream page on parallel execution). At parallelism 4 the lower bound applies: 128 key groups, held as ranges by 4 subtasks.

Two consequences. The key group of a key never changes while maximum parallelism is fixed, but the subtask owning that group changes whenever parallelism changes — that is how a job rescales without re-keying anything (module 10). And the key is not carried on the record: the keys must be produced in a deterministic way, because they are recomputed whenever they are needed, rather than being attached to the stream records. A selector that reads a clock routes the same request to different subtasks on different calls. lambda e: e.request_id is deterministic; keep it that way.

Upstream is candid about the price: every keyBy causes a network shuffle that repartitions the stream. In general this is pretty expensive, since it involves network communication along with serialization and deserialization. GatewayMeter pays it three times by module 6 — request_id, tenant_id, model — each time because the next operator needs every event for one key in one place.

Keyed state lives where the key is

Routing and state are one decision. Upstream: the state is partitioned and distributed strictly together with the streams that are read by the stateful operators. Hence, access to the key/value state is only possible on keyed streams, i.e. after a keyed/partitioned data exchange, and is restricted to the values associated with the current event's key. Aligning the keys of streams and state makes sure that all state updates are local operations, guaranteeing consistency without transaction overhead.

For the assembler: when RequestCompleted for r-4f1a arrives, the subtask handling it reads open for r-4f1a from local memory, because the hash that brought this event here brought the RequestStarted here, and the state was written here. No lookup crosses a subtask. One ValueState declaration, in the ETL guide's words, represents not just a single Boolean, but rather a distributed, sharded, key/value store, each subtask responsible for a disjoint subset of the entire keyspace. Module 3 is about what goes into it and how it is kept from growing forever.

Tasks, subtasks, slots

From the glossary, kept exact. An Operator is a node of a Logical Graph. An Operator performs a certain operation, which is usually executed by a Function. An Operator Chain consists of two or more consecutive Operators without any repartitioning in between. Operators within the same Operator Chain forward records to each other directly without going through serialization or Flink's network stack. A Task is a node of a Physical Graph. A task is the basic unit of work, which is executed by Flink's runtime. Tasks encapsulate exactly one parallel instance of an Operator or Operator Chain. A Sub-Task is a Task responsible for processing a partition of the data stream.

Chaining is the default. Upstream: Flink chains operator subtasks together into tasks. Each task is executed by one thread. Chaining operators together into tasks is a useful optimization: it reduces the overhead of thread-to-thread handover and buffering, and increases overall throughput while decreasing latency. Anything that repartitions ends a chain — a key_by, a rebalance, a parallelism change between neighbours. You override it with three calls that can only be used right after a DataStream transformation as they refer to the previous transformation: start_new_chain(), disable_chaining(), and slot_sharing_group("name") to move an operator into a different slot. The environment also has a job-wide switch, env.disable_operator_chaining() in PyFlink (disableOperatorChaining() in the Java docs). The lab uses disable_chaining() on parse once so the Web UI draws it as its own box with its own numRecordsIn and numRecordsOut; production GatewayMeter leaves it chained.

A slot is where a task's thread runs: each task slot represents a fixed subset of resources of the TaskManager — and, since slot ≠ CPU core, no CPU isolation happens here; currently slots only separate the managed memory of tasks. Slots are shared by default: Flink allows subtasks to share slots even if they are subtasks of different tasks, so long as they are from the same job. The result is that one slot may hold an entire pipeline of the job. Hence the planning rule: a Flink cluster needs exactly as many task slots as the highest parallelism used in the job.

Stage 2 under those rules, on the module-1 cluster (one JobManager, four TaskManagers with two slots each):

Four slots used, four of the cluster's eight idle. Which TaskManagers supply them is the scheduler's choice; read it off the Web UI in Lab 2.1.

Inside one of those threads, the task lifecycle page fixes the order: consecutive operators in a task are opened from the last to the first, so downstream is ready before the first operator processes input; then per element processElement(), the place where the UDF's logic is invoked; snapshotState() from another thread on each checkpoint barrier (module 7); close() on failure, skipping everything between.

The assembler, in PyFlink

Upstream's description of the tool: the ProcessFunction is a low-level stream processing operation, giving access to the basic building blocks of all (acyclic) streaming applications: events (stream elements); state (fault-tolerant, consistent, only on keyed stream); timers (event time and processing time, only on keyed stream). The ProcessFunction can be thought of as a FlatMapFunction with access to keyed state and timers. KeyedProcessFunction is the variant applied after key_by; RequestAssembler is one.

The shape matches the Python tab of the upstream process-function page: open(self, runtime_context) creates the state handle from a ValueStateDescriptor; process_element(self, value, ctx) receives one event and a context; output is produced by yield — PyFlink has no Collector argument, unlike Java. The state's type information is Types.PICKLED_BYTE_ARRAY(), which is how the upstream Python examples store a small object such as OpenRequest.

from pyflink.common.typeinfo import Types
from pyflink.datastream.functions import KeyedProcessFunction, RuntimeContext
from pyflink.datastream.state import ValueStateDescriptor

from gatewaymeter.events import Event, OpenRequest, RequestRecord, parse_event


class RequestAssembler(KeyedProcessFunction):
    """Keyed by request_id. Pairs RequestStarted with RequestCompleted."""

    def __init__(self):
        self.open_state = None  # ValueState[OpenRequest], one per request_id

    def open(self, runtime_context: RuntimeContext):
        self.open_state = runtime_context.get_state(
            ValueStateDescriptor("open", Types.PICKLED_BYTE_ARRAY()))

    def process_element(self, event: Event, ctx: "KeyedProcessFunction.Context"):
        if event.event_type == "RequestStarted":
            self.open_state.update(OpenRequest(
                tenant_id=event.tenant_id,
                model=event.model,
                started_ts=event.ts,
                prompt_tokens=event.payload["prompt_tokens"],
            ))
            return

        if event.event_type == "RequestCompleted":
            started = self.open_state.value()  # state for THIS key only
            if started is None:
                return                          # completion without a start
            self.open_state.clear()
            yield RequestRecord(
                request_id=event.request_id,
                tenant_id=started.tenant_id,
                model=started.model,
                started_ts=started.started_ts,
                completed_ts=event.ts,
                latency_ms=event.payload["latency_ms"],
                prompt_tokens=started.prompt_tokens,
                completion_tokens=event.payload["completion_tokens"],
                status=event.payload["status"],
            )
        # TokenGenerated, ToolCalled, AgentStepCompleted, ModelError: ignored at this stage


def build_job(env):
    env.set_parallelism(4)
    events = (env.from_source(gateway_events_source(), watermark_strategy(), "gateway-events")
                 .map(parse_event, output_type=Types.PICKLED_BYTE_ARRAY())
                 .name("parse"))
    records = (events
                 .key_by(lambda e: e.request_id, key_type=Types.STRING())
                 .process(RequestAssembler())
                 .name("assemble"))
    records.print()
    return env

self.open_state.value() returns the value for the key of the event being processed — the runtime sets that scope before calling you, and ctx carries the same key. ctx also exposes ctx.timestamp() and ctx.timer_service(); module 6 uses the timer service to emit status="abandoned" when a completion never comes. Until then an unpaired RequestStarted sits in open forever — module 3's failure.

The started is None branch is this module's subject. In a correctly keyed job it fires only when a completion genuinely precedes its start — rare, because one request's events share a Kafka partition and a partition is read in order, but possible when the gateway's retry buffer re-sends a RequestStarted after its RequestCompleted (module 4's late burst), or when the job restarts between the two with no checkpoint. In a mis-keyed job it fires for every request.

The failure: drop the key_by

Replace the keyed exchange with a load-spreading one — rebalance() followed by a plain ProcessFunction that keeps a Python dict of open requests, because a non-keyed stream cannot hold keyed state — and run six events through four assembler subtasks. The key groups below are illustrative hashes (the real ones are Flink's; Lab 2.4 makes them visible), the ranges are the even 32-per-subtask split of 128 groups, and the rebalance column assumes one sender dealing round-robin from subtask 0. Kafka partitions are the gateway's, assigned to source readers uniformly distributed across subtasks, in round-robin style (upstream's words for the Kafka split enumerator); the table takes partition p to reader p mod 4.

With key_by, event 3 lands on subtask 2, finds the OpenRequest that event 1 wrote there, and emits. With rebalance(), event 1 wrote its entry into the dict on subtask 0 and event 3 arrives at subtask 2, whose dict has never heard of r-4f1a. Every completion probes a different subtask from the one holding its start. Three starts sit in three dictionaries, three completions hit the is None branch, zero records reach the sink. No exception, no restart, no red box: numRecordsIn on assemble is 6 and numRecordsOut is 0. Three requests were served and none will be costed — the never-billed failure, revenue lost at the rate the gateway serves traffic.

Now the column that looks like it works. With a plain forward edge, both events of every request reach the same assembler subtask, and three records come out. That is not the key_by guarantee; it is two facts lining up. The gateway chose the Kafka partition by request_id; the split enumerator gave that partition to one source subtask; the forward edge kept the assembler's subtask equal to the source's. Nothing in the job knows request_id is the key. Raise the source parallelism to 8, or give assemble a parallelism different from the source so Flink must redistribute between them, and the co-location is gone. And while it holds, what the assembler has is a dict in a Python process: not keyed state, so not checkpointed in module 7, not moved with its key groups in module 10, invisible to a timer in module 6. Upstream's rule: state is fault-tolerant, consistent, only on keyed stream. Kafka partitioning is an input property; the Flink key is a property of the job, and the job has to declare it.

Sidebar: DataStream API v2

Upstream, as of 2026-09: DataStream API V2 is a new set of APIs, to gradually replace the original DataStream API. It is currently in the experimental stage and is not fully available for production. It reorganises this module's ideas into named "fundamental primitives" — data stream, partitioning, process function, state, processing timer service, watermark — with windows, joins and the event timer service layered on as extensions; streams are typed by their partitioning (NonKeyedPartitionStream from a source) and transformed through process. Its documentation is Java only. This course stays on the classic DataStream API because PyFlink is on it. See https://nightlies.apache.org/flink/flink-docs-release-2.3/docs/dev/datastream-v2/overview/.

If a TaskManager dies right now

Which numbers survive, which are recomputed, and which could be emitted twice? Stage 2 still has no checkpoint, so the answer is short.

Survive: only the RequestRecords already handed to the sink. The open state on the dead TaskManager's slots — every request started there and not yet completed — is gone; the job restarts with empty state or fails, depending on the restart strategy module 1 configured. Recomputed: whatever the source re-reads, and with no checkpoint there is no recorded position to rewind to, so that is a property of the source's start-offset setting, not of the state. Emitted twice: a request whose start and completion are both re-read is assembled again; one whose completion is re-read but whose start is not is never assembled — the failure above, caused by a restart instead of a mis-key. Module 3 bounds what sits in open; module 7 makes it survive.

Try it

Go to labs/02-streams-keys-and-operators/. All four exercises run stage 2 of GatewayMeter on the compose cluster, Web UI on port 8081.

• 2.1 Submit the job and read the job graph: count boxes (tasks), subtasks per box, and slots in use on the TaskManagers page; match them to the table above. Add disable_chaining() after parse, resubmit, and explain what changed in the graph and what did not change in the output.
• 2.2 Replay the module-1 event set and count RequestRecords against RequestCompleted events; account for every difference (a completion before its start, a start with no completion).
• 2.3 Break the key: replace key_by with rebalance() and the assembler with the dict version. Count emitted records. Remove the rebalance() so the edge is forward, count again, and write one sentence on why the second count is not a fix.
• 2.4 Vary parallelism (1, 2, 4, 8) and log, per event, which subtask the assembler ran on. Confirm a given request_id is stable within a run, that its subtask changes with parallelism, and that numRecordsIn per subtask is roughly even. Then set the source parallelism above the topic's 8 partitions and note what module 5 will say about idle readers.

Lab page: lab-02-streams-keys-and-operators.html — the exercises, every source file, and the run-it-locally steps.