1. Background #

This document introduces Operator Chains and object reuse in Flink, offering suggestions for performance optimization and debugging.

2. Scenario Description #

2.1 Operator Chain #

In distributed execution, Flink chains operator subtasks together to form tasks. A single thread executes each task. Chaining operators into a task is an effective optimization: it reduces overhead from thread switching and buffering, increasing overall throughput and lowering latency. Refer to the diagram below.

2.2 Object Reuse #

In streaming scenarios, the default behavior for chained operators is to perform a deep copy of data when passing it from an upstream operator to a downstream operator. This is primarily because downstream operators might reuse or modify mutable objects.

However, data copying is often costly. Therefore, Flink provides the dynamic parameter pipeline.object-reuse to control how data is passed from upstream to downstream operators within a chain.

3. Procedure #

3.1 Scenario 1: Enable Object-Reuse to Improve Performance #

3.1.1 Parameter Settings

In an existing Flink job, set pipeline.object-reuse to true.

# Enable object reuse (default: false)
pipeline.object-reuse: true

Job Development - Configure Parameters - Custom Parameter Settings:

After modifying the parameters, save the job. Then, publish and start it.

3.1.2 Parameter Verification

You can confirm if the setting has taken effect by checking the parameters in the Flink UI. If Object Reuse Mode is set to true, the configuration has taken effect.

3.2 Scenario 2: Disable Operator Chaining to Debug Performance Bottlenecks #

3.2.1 Configure Parameters

Although Operator Chaining and object reuse offer many performance advantages, they also complicate task debugging. For example, when multiple operators are chained together, it is hard to identify the specific operator causing a performance bottleneck. In such cases, you need to disable this feature.

# Disable Operator chaining (default: true)
pipeline.operator-chaining: false

Configuration path: Task Development - Configure Parameters - Custom Parameter Settings. After modifying the custom parameters, save and publish the changes.

3.2.2 Verify Parameters

You can confirm if the change has taken effect by observing the DAG. For example, the task's DAG before the restart is as follows.

After adding the parameter, the restarted task's DAG is as follows, where the originally chained Operators have reverted to their original state:

3.2.3 Notes

Note: disabling Operator Chaining is recommended for debugging scenarios only and not for production tasks.

1. The parameter changes the DAG, leading to state incompatibility and preventing recovery from a previous state.
2. Disabling operator chaining significantly degrades stream task processing performance. Use it with caution.