When does a Flink DataStream pipeline need process() with a KeyedProcessFunction rather than flatMap()?

## What process() adds Flink's documentation describes `ProcessFunction` as a low-level stream processing operation that gives access to the basic building blocks of all acyclic streaming applications: - **events** — the stream elements themselves, - **state** — fault-tolerant and consistent, only on a keyed stream, - **timers** — event time and processing time, only on a keyed stream. A useful mental model is that `ProcessFunction` is a `FlatMapFunction` with keyed state and timers bolted on. Like flatMap it is invoked once per incoming record and emits through a `Collector`. Unlike flatMap it also receives a `Context`. ```java stream.keyBy(Event::getUserId) .process(new MyKeyedProcessFunction()); ``` ## The Context is the whole difference Every call to `processElement(value, ctx, out)` receives a `Context` that gives you: **The element's event-time timestamp.** flatMap has no access to the record's timestamp at all. Any logic that compares an event's own time to something — how late it is, how far apart two events for a key are — needs this. **The `TimerService`.** From it you register callbacks at a future event-time or processing-time instant. An event-time timer's `onTimer(...)` fires when the current watermark advances to or past the timer's timestamp; a processing-time timer fires when wall-clock time reaches it. This is how you express "if no follow-up event arrives for this user within 30 minutes, emit an abandoned-session record" — logic that has no expression at all in a per-record transformation, because the interesting moment is defined by the *absence* of a record. **Side outputs.** `ctx.output(outputTag, value)` emits to a secondary stream identified by an `OutputTag`, which downstream you retrieve with `mainStream.getSideOutput(outputTag)`: ```java final OutputTag<String> rejected = new OutputTag<String>("rejected") {}; // inside processElement: ctx.output(rejected, "malformed: " + value); ``` The anonymous-subclass syntax `new OutputTag<String>("rejected"){}` is not a typo — the trailing braces preserve the generic type that erasure would otherwise remove. Side outputs are the idiomatic way to split malformed records, late records, or audit events off a pipeline without a second pass over the stream. ## Keyed versus unkeyed State and timers exist only on a keyed stream, so `ProcessFunction` on a plain `DataStream` gives you events, timestamps and side outputs but no keyed state and no keyed timers. Applying it after `keyBy()` unlocks both. `KeyedProcessFunction<K, IN, OUT>` is the keyed counterpart — a sibling class rather than a subclass, since both extend `AbstractRichFunction` — and its contexts additionally expose the current key via `ctx.getCurrentKey()`, which matters most inside `onTimer(...)`, because a timer callback carries no record to read the key from. ## Timer semantics worth knowing The `TimerService` **deduplicates timers per key and timestamp**: at most one timer exists for a given key and timestamp, so registering the same timestamp repeatedly does not multiply callbacks and `onTimer()` is invoked once. This is the basis of *timer coalescing* — rounding registration timestamps to a coarser granularity (say, the next full second) deliberately collapses many near-identical timers into one, which is a standard way to cut timer volume in a high-cardinality job. Flink also **synchronizes invocations of `onTimer()` and `processElement()`**, so user code never has to reason about concurrent modification of state between the two. During an `onTimer` call the state is scoped to the key the timer was created with, which is what lets a timer manipulate that key's state directly. Timers are checkpointed along with state, so they survive failure and restart. ## When flatMap is the right answer Do not reach for `process()` reflexively. If the logic is genuinely per-record — parse, validate, explode a list field, filter and reshape — `flatMap()` says so more clearly and costs less. Process functions with state carry real operational weight: state grows with key cardinality, must be checkpointed, and needs a TTL or an explicit clearing path or it grows without bound. Registering a timer per record on a high-cardinality key space produces a timer set that can rival the state itself in size. The honest decision rule is: does the output for this record depend on anything other than this record, or does something need to happen at a time when no record arrives? If neither, flatMap. If either, process. ## The escape hatch below it Below `ProcessFunction` sits the custom-operator API, but the documentation is explicit that custom operators are an advanced usage pattern and that for most use cases you should consider a process function instead. Custom operators must respect assumptions that differ between STREAMING and BATCH execution — for instance, in BATCH mode records are processed key by key and the watermark switches from `MAX_VALUE` back to `MIN_VALUE` between keys, so an operator that caches the last seen watermark and assumes it only ascends will produce wrong results. Process functions are insulated from most of that.