...
- If a Transformation has isOutputOnEOF == true:
- The process of operator chain can still be done. After that, the results of its operator chain as well as and its upstream operators have will be set blocking (by default) or h
ExecutionOptions.BATCH_SHUFFLE_MODE.
- This operator as well as its upstream operators will be executed in batch mode (e.g checkpoint is disabled when these operators are running).
- The process of operator chain can still be done. After that, the results of its operator chain as well as and its upstream operators have will be set blocking (by default) or h
- If all Transformation has isOutputOnCheckpoint == false:
- In FLIP-325, JM will not trigger an extra flush() before triggering a checkpoint.
...
5 times in 4 kinds of configuration: streaming mode, batch mode, optimized streaming mode after this PR and optimized streaming mode with hybrid shuffle after this PR. Here are the throughput benchmark results:
STREAMING | BATCH | Optimized STREAMING | With hybrid shuffle | |
---|---|---|---|---|
8e7 | 66 ± 1 (100%, 1202426 ms) | 491 ± 5 (743%, 162731 ms) | 1506 ± 10 (2281%, 53098 ms) | 1677 ± 42 (2540%, 47687 ms) |
This shows with the changes proposed above, DataStream#coGroup in Optimized STREAMING can be 20X faster than STREAMING throughput and 3X faster than BATCH
Benchmark: Aggregate
Data Count
...
5 times in 4 kinds of configuration: streaming mode, batch mode, optimized streaming mode after this PR and optimized streaming mode with hybrid shuffle after this PR. Here are the throughput benchmark results:
STREAMING | BATCH | Optimized STREAMING | With hybrid shuffle | |
---|---|---|---|---|
8e7 | 163 ± 0 (100%, 490478 ms) | 1561 ± 16 (957%, 51237 ms) | 1733 ± 9 (1063%, 46143 ms) | 1992 ± 15 (1222%, 40148 ms) |
This shows with the changes proposed above, DataStream#aggregate in Optimized STREAMING can be 10X faster than STREAMING throughput and slightly faster to the BATCH
Benchmark: OperatorDelayedDeploy
...
Code Block | ||
---|---|---|
| ||
taskmanager.numberOfTaskSlots: 2 |
Source1
,Process1
, Source2
and Process2
in 4 different SlotSharingGroups as above and then run this job, we could see that, before this PR, all of the tasks couldn't deploy and blocked in CREATED
state because only 2 slots provided in TaskManager. With our work, Source2
and Process2
can be deplyed after Source1
,Process1
finished and released their slots.
Without setting the slot sharing group, we run job in STREAMING, Optimized STREAMING and Optimized STREAMING With hybrid shuffle. As the job records number grows, throughput in STREAMING gets decreased because of the buffer presure in Process2
, and even causes OOM exception while it works well in Optimized STREAMING. Using hybrid shuffle can make Process1
to pipeline consume records from Source1
Compatibility, Deprecation, and Migration Plan
...