...

8) Hybrid shuffle can be applied by configuring ExecutionOptions.BATCH_SHUFFLE_MODE to determine the result partition type of isOutputOnEOF operator and its upstream operators in RuntimeExecutionMode.STREAMING.

Benchmark results

...

Use DataStream#coGroup to process records from two bounded streams with Data number of records and emit results after both inputs have ended.

...

With this PR, when DataStream#coGroup is invoked with EndOfStreamWindows as the window assigner, EOFAggregationOperator will be applied to sort two inputs first, and then output the coGroup results.

data1.coGroup(data2)
       .where(tuple -> tuple.f0)
       .equalTo(tuple -> tuple.f0)
       .window(EndOfStreamWindows.get())
       .apply(new CustomCoGroupFunction())
       .addSink(...);

We run

...

Here are the benchmark results:

...

each benchmark in CoGroupDataStream 5 times, here are the throughput benchmark results:

...

This shows with the changes proposed above, DataStream#coGroup can be

...

15~30X faster than its current throughput in stream mode, and 2.5X faster than its current throughput in batch mode. Using hybrid shuffle in the optimized non-pipeline part can further improve the performance.

Compatibility, Deprecation, and Migration Plan

...