Understanding Flink Operator Chaining Mechanism

Many users encounter a Flink Web UI that shows only a single box with zero records sent and received, which is often caused by the operator chain mechanism rather than bugs in the program.

Flink builds its execution plan using three graph layers: StreamGraph (original logical plan), JobGraph (optimized logical plan shown in the UI), and ExecutionGraph (physical plan). The operator chain is added during the transformation from StreamGraph to JobGraph.

The core method private JobGraph createJobGraph() computes hashes for each node, prepares a map for chained operator hashes, and calls setChaining(), which iterates over source nodes and invokes createChain(). This recursive method builds lists of transitive output edges, chainable outputs, and non‑chainable outputs, determines chaining eligibility, and configures StreamConfig objects for each operator.

Chaining eligibility is defined in

public static boolean isChainable(StreamEdge edge, StreamGraph streamGraph)

. An upstream and downstream operator can be chained only if they share the same slot sharing group, both have appropriate chaining strategies (ALWAYS for downstream, HEAD or ALWAYS for upstream), use a forward partitioner, are not in batch shuffle mode, have equal parallelism, and chaining is globally enabled.

Users can disable chaining for a specific operator with disableChaining() or start a new chain with startNewChain(). To disable chaining globally, call StreamExecutionEnvironment.disableOperatorChaining().

At runtime, each TaskManager creates an OperatorChain inside a StreamTask. The chain consists of a head operator, an array of all operators (in reverse order), and stream outputs. The method createOutputCollector() builds collectors for both network outputs and chained outputs, recursively creating chained operators via createChainedOperator(). The ChainingOutput.collect() method forwards records directly to the next operator's processElement(), eliminating extra overhead.

Overall, operator chaining groups compatible operators into a single thread (TaskManager slot), reducing data exchange, serialization, and context switches, thereby improving execution efficiency.