...

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Currently, Flink allows the same operator to be used in both stream mode and batch mode. The operator can process bounded stream of records with high throughput (e.g. via sorting/buffering) in a batch-mode job, and process unbounded stream of records with low processing latency (e.g. via checkpoint with stateful state backend) in a batch-mode job. However, the operator is not able to use different modes (i.e. batch and stream modes) at different stages of the same job, which makes it hard to meet the performance requirement for jobs that need to process a bounded stream of backlog data followed by an unbounded stream of fresh data.

The bounded/unbounded streams of records can either come from the same source or come from different sources, as shown in the following examples:

1) We might have a job that aggregates records from a HybridSource (composed of FileSource and KafkaSource) and emits the results to Hudi Sink. When the job is processing records from the FileSource, user needs the job to maximize throughput without having to emit intermediate results with low processing latency. When the job is processing records from the KafkaSource, user needs the job to emit results with low processing latency.

2) As shown in the figure below, we might have a job that bootstraps its state (operatorA and operatorB) using records from a bounded source (i.e. inputA). There is no need to emit any intermediate result when the job is processing records from the bounded source. After all records from the bounded source have been processed by operatorA and operatorB, the job needs to process records from the unbounded source and emit results with low processing latency in real-time.

Currently, supporting the above use-cases requires all operators to run in the stream mode across the entire job lifetime. This approach leads to inferior performance because some operators (e.g. co-group) might be more than 10X slower in stream mode than when it is run in batch mode, due to lack of the capability to do buffering and sorting.

In this FLIP, we propose to optimize performance for the above use-cases by allowing an operator to effectively switch its execution mode from batch to stream based on the "backlog" status of the records emitted by the source operators.

NOTE: This FLIP focuses only on the capability to switch from batch to stream mode. If there is any extra API needed to support switching from stream to batch mode, we will discuss them in a follow-up FLIP.

Image Removed

Behavior changes when switching from batch mode to stream mode

In this section, we describe how batch mode and stream mode differ in a variety of aspects. We also describe the new behavior needed for an operator to switch from batch mode to stream mode during execution of the same job.

...

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

[This FLIP proposal is a joint work between Xuannan Su  and Dong Lin ]

Motivation

Currently, Flink allows the same operator to be used in both stream mode and batch mode. The operator can process bounded stream of records with high throughput (e.g. via sorting/buffering) in a batch-mode job, and process unbounded stream of records with low processing latency (e.g. via checkpoint with stateful state backend) in a batch-mode job. However, the operator is not able to use different modes (i.e. batch and stream modes) at different stages of the same job, which makes it hard to meet the performance requirement for jobs that need to process a bounded stream of backlog data followed by an unbounded stream of fresh data.

The bounded/unbounded streams of records can either come from the same source or come from different sources, as shown in the following examples:

1) We might have a job that aggregates records from a HybridSource (composed of FileSource and KafkaSource) and emits the results to Hudi Sink. When the job is processing records from the FileSource, user needs the job to maximize throughput without having to emit intermediate results with low processing latency. When the job is processing records from the KafkaSource, user needs the job to emit results with low processing latency.

2) As shown in the figure below, we might have a job that bootstraps its state (operatorA and operatorB) using records from a bounded source (i.e. inputA). There is no need to emit any intermediate result when the job is processing records from the bounded source. After all records from the bounded source have been processed by operatorA and operatorB, the job needs to process records from the unbounded source and emit results with low processing latency in real-time.

Currently, supporting the above use-cases requires all operators to run in the stream mode across the entire job lifetime. This approach leads to inferior performance because some operators (e.g. co-group) might be more than 10X slower in stream mode than when it is run in batch mode, due to lack of the capability to do buffering and sorting.

In this FLIP, we propose to optimize performance for the above use-cases by allowing an operator to effectively switch its execution mode from batch to stream based on the "backlog" status of the records emitted by the source operators.

NOTE: This FLIP focuses only on the capability to switch from batch to stream mode. If there is any extra API needed to support switching from stream to batch mode, we will discuss them in a follow-up FLIP.

Image Added

Behavior changes when switching from batch mode to stream mode

In this section, we describe how batch mode and stream mode differ in a variety of aspects. We also describe the new behavior of Flink runtime which supports an operator to switch from batch mode to stream mode during execution of the same job.

For easy of understanding, we use a simplified use-case with the following properties when describing the behavior changes:

• The job uses default configuration values for all configurations (e.g. scheduler-mode) except execution.runtime-mode (see below).
• The job has one source that switches from isBacklog=true to isBacklog=false before it reaches EOF.

Here are the definition of batch mode, stream mode and mixed mode used below:

• Batch mode refers to the behavior of Flink runtime prior to this FLIP with execution.runtime-mode = batch.
• Stream mode refers to the behavior of Flink runtime prior to this FLIP with execution.runtime-mode = streaming.
• Mixed mode refers to the behavior of Flink runtime after this FLIP with execution.runtime-mode = streaming AND execution.checkpointing.interval-during-backlog = 0.

And we make the following extra notes for the behavior changes:

• After this FLIP, the behavior of Flink runtime with execution.runtime-mode = streaming AND execution.checkpointing.interval-during-backlog > 0, will be same as the stream mode prior to this FLIP.
• The rational for switching the Flink runtime behavior based on whether execution.checkpointing.interval-during-backlog == 0 is that most (if not all) performance optimizations, which allow batch mode to achieve higher throughput than stream mode, involve operations (e.g. sorting inputs) that can only be used when checkpoint is not required.
• It is possible for mixed mode to be slower than stream mode, particularly when there is only small amount of input records and the overhead of buffering/sorting inputs out-weight its benefit. This is similar to how the merge join might be slower than hash join. This FLIP focuses on optimizing the Flink throughput when there is a high number of input records. In the future, we might introduce more strategies to turn on mix mode in a smart way to avoid performance regression.
• For an operator with 2+ inputs, where some inputs have isBacklog=true and some other inputs have isBacklog=false, Flink runtime will handle this operator as if all its inputs have isBacklog=false. The rational is that we don't have a reliable way (yet) to decide whether it is OK to delay the processing/emission of this operator's output records. For example, if this operator simply forwards the records from the input with backlog=false to its output, then we probably should not delay its output records.

1) Scheduling strategy

Batch mode:

...

• All tasks are deployed at the beginning of the job without waiting for upstream tasks to finish.

...

Mixed mode:

• Same as stream mode

...

• .

The capability to update/optimize scheduling (e.g. speculative execution, AQE) in mixed mode will be left to future work.

...

• All shuffle type will be set to PIPELINED.

Mixed mode with execution.checkpointing.interval-during-backlog > 0:

...

• .

Mixed mode with execution.checkpointing.interval-during-backlog = 0:

• In mixed mode, the same shuffle strategy as stream mode will be used to set shuffle type between tasks.
• Before source operator emits isBacklog=false, keyed input of one input operator will be automatically sorted during isBacklog=true if it doesn't sort the input internally.
  • Keyed inputs of multiple inputs operator are not automatically sorted. It can sort the inputs internally.
• When isBacklog switches to false, the keyed inputs will not be sorted.

...

• Source operator emits watermark based on the user-specified WatermarkStrategy and pipeline.auto-watermark-interval while it has not reached EOF.
• Once the source operator reaches EOF, it emits watermark of Long.MAX_VALUE after it has emitted all records. That means the downstream operator will not see any further input records after having received watermark=Long.MAX_VALUE.

Mixed mode with execution.checkpointing.interval-during-backlog > 0:

...

• .MAX_VALUE.

Mixed mode with execution.checkpointing.interval-during-backlog = 0:

• Source operator does not emit watermark while isBacklog=true.

• At the point when isBacklog switches to false, source operator emits RecordAttributes(isBacklog=false) and the greatest watermark during backlog processing.

• Source operator emits watermark based on the user-specified WatermarkStrategy and pipeline.auto-watermark-interval while it has not reached EOF.

• Once the source operator reaches EOF, it emits watermark of Long.MAX_VALUE after it has emitted all records.

...

• Every operator needs to support checkpoint. Checkpoint is triggered periodically according to execution.checkpointing.interval.checkpointing.interval.
• If any task fails, its pipelined region is restarted to re-process its input since the last successful checkpoint.

Mixed mode:

• Every operator needs to support checkpoint.
• Before source operator emits isBacklog=false, checkpoint triggering is disabled.
• If any task fails , its pipelined region when isBacklog=true, this task is restarted to re-process its input since from the last successful checkpoint.

Mixed mode with execution.checkpointing.interval-during-backlog > 0:

• Same as stream mode.

Mixed mode with execution.checkpointing.interval-during-backlog = 0:

• beginning.
• At the point when isBacklog switches to false, source operator emits RecordAttributes(isBacklog=false) and triggers an immediate checkpoint.
• Checkpoint is triggered periodically according to execution.checkpointing.interval
• Every operator needs to support checkpoint.
• Before source operator emits isBacklog=false, checkpoint triggering is disabled.
• If any task fails when isBacklog=true, this task false, its pipelined region is restarted to re-process its input from since the beginning.
• At the point when isBacklog switches to false, source operator emits RecordAttributes(isBacklog=false) and triggers an immediate checkpoint.
Checkpoint is triggered periodically according to
• last successful checkpoint.

Extra notes: For jobs with multiple sources and execution.checkpointing.interval-during-backlog = 0, checkpoint triggering is enabled if and only if all sources have isBacklog=false. (More details for the checkpointing behavior can be found in the doc of execution.checkpointing.interval

...

-during-backlog).

As a result, suppose a source has isBacklog=true, and another source switches from isBacklog=true to isBacklog=false,

...

the job's checkpoint would still be disabled

5) Keyed state backend

Batch mode:

...

• Managed memory is allocated to operators if the keyed state backend is a subclass of AbstractManagedMemoryStateBackend.
• A general purpose keyed state backend which does not assume inputs sorted by key (e.g. EmbeddedRocksDBStateBackend) is used.

Mixed mode with execution.checkpointing.interval-during-backlog > 0:

...

Mixed mode with execution.checkpointing.interval-during-backlog = 0:

• Managed memory is allocated to operators with keyed inputs.

• A general purpose keyed state backend that does not assume inputs sorted by key (e.g. EmbeddedRocksDBStateBackend) is instantiated by Flink runtime.

• Before source operator emits isBacklog=false (during backlog processing), the keyed input to one input operator is sorted by the key.

  • Multiple input operators Operators with multiple inputs can buffer and sort the inputs internally. We will provide utility classes (derived from the existing ExternalSorter) to sort inputs.
    ExternalSorter) to sort inputs.
    
  • Flink runtime will buffer/aggregate state's key/value in memory before persisting state's key/value to the underlying state backend (e.g. rocksdb), such that for the input records that are already sorted on the given keyWith the sorted input and the state backend optimization introduced in FLIP-325, there will be at most one read/write access to the Rocksdb underlying state backend for each key. And we only need to write to the Rocksdb state backend if the lru cache is full.
• At the point when isBacklog switches to false:
  • Source operator emits RecordAttributes(isBacklog=false).
• The operator continues to process records using the keyed state backend instantiated by Flink runtime, which now contains the full state obtained from records received while isBacklog=true.

...

• The timer service keeps timers for all the keys, and it will fire timers based on the watermark.

Mixed mode with execution.checkpointing.interval-during-backlog > 0:

...

• watermark.

Mixed mode with execution.checkpointing.interval-during-backlog = 0:

• Before the source operator emits isBacklog=false, the timer service would fire processing-time timers based on the system time.
• At the point when isBacklog switches to false, the timer service would fire event-time timers up to the latest watermark at this point keeps timers only for the current key, and it will fire timers of the current key up to the last watermark during backlog at the end of each key. And all the processing time timer up to the current system time will be triggered.
  
  • The value of InternalTimeService#getCurrentWatermark will be Watermark.MIN_VALUE when processing a key and will be set to the last watermark during backlog when firing triggers.
• After isBacklog switches to false, the timer service keeps timers for all the keys and fires timers as the watermark advancescontinues to fire processing-time and even-time timers in the same way as the stream mode.

Public Interfaces

1) Add RecordAttributesBuilder and RecordAttributes that extends RuntimeEvent to StreamElement to provide operator with essential information about the records they receive, such as whether the records are already stale due to backlog.

...

/**
 * A RecordAttributes element provides stream task with information that can be used to optimize
 * the stream task's performance.
 */
@Experimental
public class RecordAttributes extends RuntimeEventStreamElement {
    /**
     * If it returns true, then the records received after this element are stale
     * and an operator can optionally buffer records until isBacklog=false. This
     * allows an operator to optimize throughput at the cost of processing latency.
     */
     @Nullable
     public Boolean isBacklog() {...}
}

...

@PublicEvolving
public interface Input<IN> {
    ...

    /**
     * Processes a {@link RecordAttributes} that arrived on this input.
     * This method is guaranteed to not be called concurrently with other methods of the operator is guaranteed to not be called concurrently with other methods of the operator.
	 * The recordAttributes do not need to be persisted in the checkpoint, as the source generates
	 * RecordAttributes after recovery from checkpoint.
     */
    @Experimental
    default void processRecordAttributes(RecordAttributes recordAttributes) throws Exception {}
}

...

@PublicEvolving
public interface TwoInputStreamOperator<IN1, IN2, OUT> extends StreamOperator<OUT> {
    ...

    /**
     * Processes a {@link RecordAttributes} that arrived on the first input of this operator.
     * This method is guaranteed to not be called concurrently with other methods of the operator.
 	 * The recordAttributes do not need to not be called concurrently with other methods of the operator.
      be persisted in the checkpoint, as the source generates
	 * RecordAttributes after recovery from checkpoint. 
     */
    @Experimental
    default void processRecordAttributes1(RecordAttributes recordAttributes) throws Exception {}

    /**
     * Processes a {@link RecordAttributes} that arrived on the second input of this operator.
     * This method is guaranteed to not be called concurrently with other methods of the operator. other methods of the operator.
 	 * The recordAttributes do not need to be persisted in the checkpoint, as the source generates
	 * RecordAttributes after recovery from checkpoint. 
     */
    @Experimental
    default void processRecordAttributes2(RecordAttributes recordAttributes) throws Exception {}
}

...

) throws Exception {}
}

4) Add OperatorAttributesBuilder and OperatorAttributes for operator developers to specify operator attributes that Flink JM can use to properly compile the graph (e.g. whether to add an external sorter to sort input records by key).

...

• If a OneInputTransformation has isInternalSorterSupported == false, keyed input, and execution.checkpointing.interval-during-backlog == 0:
  • Flink runtime will add an external sorter for the input to sort the keyed input during backlog processing
  • Its managed memory should be set according to execution.sorted-inputs.memory
  • Flink runtime will buffer/aggregate state's key/value in memory before persisting state's key/value to the underlying state backend (e.g. rocksdb), such that for the input records that are already sorted on the given key, there will be at most one read/write access to the underlying state backend for each key.

With the above change, we In addition to the above change, we will be making use of the state backend optimization introduced in FLIP-325. We will use a state backend with a cache size of 1 if the cache size isn't explicitly specified by users so that we optimize the state backend access while minimizing the memory overhead. Alternatively, when users have set a cache size, we will adhere to that setting. This optimization can significantly enhance the performance of keyed one input operators without code change when "isBacklog=true". Moreover, the performance of the operator during isBacklog=true is close to the performance in batch mode.

65) For keyed multi-input operators that involve aggregation operation (e.g. join, cogroup, aggregate), update it to take advantage of the sorted input when isBacklog=true.

...

• Override getOperatorAttributes to return IsInternalSorterSupported = true.
• Override processRecordAttributes() API to adjust its behavior based on the input's isBacklog status.
  • The operator can buffer and sort input records when any input has isBacklog=true AND execution.checkpointing.interval-during-backlog=0.
  • Once all inputs' status has switched to isBacklog=false, it processes the buffered records, emits results, and starts to work in the stream execution mode.

Evaluation

Analysis of APIs affected by this FLIP

1) After the proposed changes, the following DataStream API should have similar performance as batch mode during backlog processing.

...

ConnectedStreams:

• ConnectedStreams#process

2) After the proposed change, the following keyed one input operations (not exhaustive) in SQL API will have a similar performance as batch mode during backlog processing:

• Window Aggregation
• Group Aggregation
• Over Aggregation

For muti-input operations, such as Regular Join, Interval Join, and Temporal Join, can be updated gradually to optimize the performance during backlog processing.

Benchmark results

In this section, we provide benchmark results to compare the throughput between batch mode and stream mode with and without backlog processing optimization for commonly-used operations during backlog processing. This will demonstrate the performance we can gain with the optimization during backlog processing.

...