Definition
- ProcessFunction – Flink's lowest-level stream-processing primitive that gives per-element control over the stream with access to context, timers, and side outputs.
- KeyedProcessFunction – The keyed variant used on a keyed stream, adding access to the current key and keyed state scoped automatically to that key.
- Timers – Event-time or processing-time callbacks registered via
ctx.timerService().registerEventTimeTimer(t)that fire anonTimermethod when the time is reached. - Keyed state – State handles like ValueState, ListState, and MapState declared in
open()and scoped to the current key, giving each account its own independent data. - Side outputs – Secondary typed streams emitted via
ctx.output(outputTag, value)to split main results from alerts or anomalies.
The high-level window operators at the National Bank of India are like pre-printed forms: a form for hourly totals, a form for sliding averages. They are convenient when your need fits the form. ProcessFunction is like giving an experienced officer a blank sheet and full authority: for each transaction they can look at it, remember things about this specific customer in a personal ledger (keyed state), set a reminder alarm for later (timer), and drop unusual items into a separate review tray (side output). When the bank needs a custom rule that no pre-printed form covers, you hand the officer the blank sheet.
Understanding the Concepts
What is ProcessFunction in Flink?
A ProcessFunction is applied with stream.process(new MyProcessFunction()). Flink calls the processElement method once per incoming element, giving you the element, a Context, and a Collector to emit output.
package com.bank.streaming;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import java.math.BigDecimal;
public class TransactionRouter
extends ProcessFunction<Transaction, Transaction> {
@Override
public void processElement(Transaction txn, Context ctx,
Collector<Transaction> out) throws Exception {
// Access element timestamp
long timestamp = ctx.timestamp();
// Access current watermark
long watermark = ctx.timerService().currentWatermark();
// Emit to main output
if (txn.getAmount().compareTo(new BigDecimal("50000")) > 0) {
out.collect(txn); // high-value transaction
}
}
}
What is KeyedProcessFunction?
On a keyed stream, you use KeyedProcessFunction which adds access to the current key and keyed state. Each key (each account) gets its own independent state and timers.
package com.bank.streaming;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import java.math.BigDecimal;
public class AccountBalanceMonitor
extends KeyedProcessFunction<String, Transaction, String> {
private transient ValueState<BigDecimal> dailyTotal;
@Override
public void open(Configuration parameters) {
dailyTotal = getRuntimeContext().getState(
new ValueStateDescriptor<>("dailyTotal", BigDecimal.class));
}
@Override
public void processElement(Transaction txn, Context ctx,
Collector<String> out) throws Exception {
String accountId = ctx.getCurrentKey(); // current key
BigDecimal current = dailyTotal.value();
if (current == null) current = BigDecimal.ZERO;
BigDecimal updated = current.add(txn.getAmount());
dailyTotal.update(updated);
if (updated.compareTo(new BigDecimal("500000")) > 0) {
out.collect("DAILY LIMIT ALERT: " + accountId
+ " total=Rs" + updated);
}
}
}
Usage:
stream.keyBy(Transaction::getAccountId)
.process(new AccountBalanceMonitor())
.print("Alerts");
Output:
Alerts> DAILY LIMIT ALERT: ACC-001 total=Rs520000
Alerts> DAILY LIMIT ALERT: ACC-003 total=Rs750000
How do timers work in KeyedProcessFunction?
Timers are registered per key. When event time (or processing time) reaches the timer's time, Flink calls onTimer. This is how you build timeouts, session expirations, and custom triggers.
package com.bank.fraud;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
public class LoginTimeoutDetector
extends KeyedProcessFunction<String, LoginEvent, String> {
private transient ValueState<LoginEvent> pendingLogin;
private static final long TIMEOUT_MS = 5 * 60 * 1000; // 5 minutes
@Override
public void open(Configuration parameters) {
pendingLogin = getRuntimeContext().getState(
new ValueStateDescriptor<>("pendingLogin", LoginEvent.class));
}
@Override
public void processElement(LoginEvent event, Context ctx,
Collector<String> out) throws Exception {
if (event.getType().equals("LOGIN")) {
// Store login and register timer
pendingLogin.update(event);
long timerTime = event.getTimestamp() + TIMEOUT_MS;
ctx.timerService().registerEventTimeTimer(timerTime);
} else if (event.getType().equals("TRANSACTION")) {
// Activity detected: cancel timeout
if (pendingLogin.value() != null) {
long timerTime =
pendingLogin.value().getTimestamp() + TIMEOUT_MS;
ctx.timerService().deleteEventTimeTimer(timerTime);
pendingLogin.clear(); // clean up state
}
}
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx,
Collector<String> out) throws Exception {
// Timer fired: no transaction within 5 minutes
LoginEvent login = pendingLogin.value();
if (login != null) {
out.collect("STALE SESSION ALERT: " + ctx.getCurrentKey()
+ " logged in at " + login.getTimestamp()
+ " but no activity within 5 minutes");
pendingLogin.clear(); // clean up
}
}
}
Timer flow for account ACC-001:
10:00:00 LOGIN event arrives
Store login, register timer for 10:05:00
10:03:00 TRANSACTION event arrives
Cancel timer, clear state (session active)
Timer flow for account ACC-002:
10:00:00 LOGIN event arrives
Store login, register timer for 10:05:00
10:05:00 Timer fires (no transaction arrived)
Output: STALE SESSION ALERT: ACC-002
How do side outputs work?
Side outputs let you split a stream into multiple typed outputs. You define an OutputTag for each secondary stream and emit to it via ctx.output(). The main stream and side streams are separate and can have different types.
package com.bank.streaming;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import java.math.BigDecimal;
public class TransactionClassifier
extends KeyedProcessFunction<String, Transaction, Transaction> {
// Side output for suspicious transactions
public static final OutputTag<Transaction> SUSPICIOUS_TAG =
new OutputTag<Transaction>("suspicious") {};
// Side output for high-value transactions
public static final OutputTag<Transaction> HIGH_VALUE_TAG =
new OutputTag<Transaction>("high-value") {};
@Override
public void processElement(Transaction txn, Context ctx,
Collector<Transaction> out) throws Exception {
BigDecimal amount = txn.getAmount();
if (txn.isFlagged()) {
ctx.output(SUSPICIOUS_TAG, txn); // side output
} else if (amount.compareTo(new BigDecimal("100000")) > 0) {
ctx.output(HIGH_VALUE_TAG, txn); // side output
} else {
out.collect(txn); // main output
}
}
}
// Usage:
// SingleOutputStreamOperator<Transaction> mainStream =
// stream.keyBy(Transaction::getAccountId)
// .process(new TransactionClassifier());
//
// DataStream<Transaction> suspicious =
// mainStream.getSideOutput(TransactionClassifier.SUSPICIOUS_TAG);
// DataStream<Transaction> highValue =
// mainStream.getSideOutput(TransactionClassifier.HIGH_VALUE_TAG);
How is ProcessFunction different from window operators?
| Feature | Window Operators | ProcessFunction |
|---|---|---|
| Paradigm | Declarative: define window + aggregation | Imperative: per-element control |
| State management | Automatic (window state) | Manual (explicit state handles) |
| Timer control | Implicit (window fires on watermark) | Explicit (register, cancel, fire) |
| Triggering | Window trigger (count, time, custom) | Custom: any condition |
| Side outputs | Limited | Full support |
| Complexity | Low (fits standard patterns) | High (you control everything) |
| Use when | Standard tumbling/sliding/session windows | Custom stateful timer logic |
// Window operator: simple tumbling sum
stream.keyBy(Transaction::getAccountId)
.window(TumblingEventTimeWindows.of(Time.minutes(5)))
.sum("amount"); // simple, automatic
// ProcessFunction: custom timeout detection
// No window operator can express "alert if login is not
// followed by a transaction within 5 minutes per account"
stream.keyBy(LoginEvent::getAccountId)
.process(new LoginTimeoutDetector()); // custom, manual
What are common mistakes with ProcessFunction?
The most common mistake is forgetting to clean up per-key state and timers. If state is never cleared, it grows unboundedly as new keys appear, eventually exhausting memory or disk. Another mistake is using processing-time timers when event-time timers are needed for deterministic replay.
// WRONG: state never cleaned up
public void processElement(Transaction txn, Context ctx,
Collector<String> out) throws Exception {
BigDecimal current = balance.value();
if (current == null) current = BigDecimal.ZERO;
balance.update(current.add(txn.getAmount()));
// State grows forever for every key that ever appeared!
}
// CORRECT: clean up state when no longer needed
public void processElement(Transaction txn, Context ctx,
Collector<String> out) throws Exception {
BigDecimal current = balance.value();
if (current == null) current = BigDecimal.ZERO;
BigDecimal updated = current.add(txn.getAmount());
balance.update(updated);
// Register cleanup timer (state TTL alternative)
ctx.timerService().registerEventTimeTimer(
ctx.timestamp() + 24 * 60 * 60 * 1000); // 24h cleanup
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx,
Collector<String> out) throws Exception {
balance.clear(); // clean up old state
}
When should you use ProcessFunction vs window operators?
Use window operators for standard tumbling, sliding, or session windows with aggregations. Drop to ProcessFunction for custom triggering, cross-event conditional logic, data-dependent timeouts, or patterns where the firing rule is business-specific.
Decision guide:
Tumbling/sliding/session window aggregation? --> Window operator
Standard count or time trigger? --> Window operator
Custom timeout per key? --> ProcessFunction
Cross-event conditional logic? --> ProcessFunction
State machine with transitions? --> ProcessFunction
Need side outputs for splitting? --> ProcessFunction
Business-specific firing rules? --> ProcessFunction
What should a senior engineer know about ProcessFunction?
A senior engineer should prefer event-time timers over processing-time timers for correctness because event-time timers fire deterministically on replay. They should use state TTL as an automatic cleanup mechanism, understand that each registered timer consumes memory, and know that ProcessFunction is the foundation on which Flink CEP is built.
package com.bank.streaming;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
public class SeniorProcessFunctionConfig {
// Use state TTL for automatic cleanup instead of manual timers
public static ValueStateDescriptor<BigDecimal> createTtlState() {
ValueStateDescriptor<BigDecimal> desc =
new ValueStateDescriptor<>("balance", BigDecimal.class);
StateTtlConfig ttlConfig = StateTtlConfig
.newBuilder(Time.hours(24))
.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
.setStateVisibility(
StateTtlConfig.StateVisibility.NeverReturnExpired)
.build();
desc.enableTimeToLive(ttlConfig);
return desc;
}
}
Visual Overview
KeyedProcessFunction Capabilities
┌───────────────────────────────────────────────┐
│ KeyedProcessFunction │
│ │
│ processElement(element, ctx, collector) │
│ │ │
│ ├── ctx.getCurrentKey() --> "ACC-001" │
│ ├── ctx.timestamp() --> event time │
│ ├── ctx.timerService() │
│ │ ├── currentWatermark() │
│ │ ├── registerEventTimeTimer(t) │
│ │ └── deleteEventTimeTimer(t) │
│ ├── ctx.output(tag, value) --> side output │
│ └── collector.collect(val) --> main output │
│ │
│ onTimer(timestamp, ctx, collector) │
│ │ │
│ └── fires when timer time is reached │
│ │
│ Keyed State (per key): │
│ ValueState<BigDecimal> balance │
│ ListState<Transaction> history │
│ MapState<String, Alert> alerts │
└───────────────────────────────────────────────┘
Login Timeout State Machine
┌─────────┐ LOGIN ┌──────────────┐ TRANSACTION ┌─────────┐
│ IDLE │────────▶│ WAITING │──────────────▶│ ACTIVE │
│ │ │ (timer set) │ │(cleared)│
└─────────┘ └──────┬───────┘ └─────────┘
│ timer fires
▼
┌──────────────┐
│ ALERT EMITTED│
│ (state cleared)│
└──────────────┘
Key Takeaway
ProcessFunction and KeyedProcessFunction are Flink's lowest-level stream-processing primitives, giving complete per-element control. In processElement you receive each element along with a Context exposing the current key, the element's timestamp, the current watermark, a TimerService to register event-time or processing-time timers, and the ability to emit to side outputs. A companion onTimer callback fires when a registered timer's time is reached.
You reach for ProcessFunction when the standard window operators run out of road. Tumbling, sliding, and session windows with ordinary aggregations are best served by the high-level API. But when you need custom triggering, cross-event conditional logic, data-dependent timeouts, or business-specific firing rules, you drop to ProcessFunction because you control exactly when to emit, when to set and cancel timers, and what to store.
At the National Bank of India, I used a KeyedProcessFunction to detect stalled sessions: an account that logged in but had no transaction within five minutes. No standard window operator expressed that cleanly. On each login I stored the event and registered an event-time timer for five minutes later; if a transaction arrived I cancelled the timer and cleared state; if the timer fired I emitted a timeout alert to a side output. I was careful to clear per-account state whenever a session resolved, so state stayed bounded.
The senior points are to prefer event-time timers for deterministic replay, always clean up state and timers to prevent unbounded growth, and use state TTL as an automatic cleanup mechanism.
Remember one sentence: ProcessFunction gives per-element access to keyed state, timers, and side outputs for custom stateful logic that window operators cannot express.
---