jto opened a new issue, #40607: URL: https://github.com/apache/beam/issues/40607
### What would you like to happen? While running large Scio batch jobs on the Flink runner (Flink 1.20 and 2.2), profiling turned up several places where the runner spends a large share of the CPU on overhead rather than on user code, or spreads work unevenly across workers. This issue tracks a set of independent improvements, each in its own PR. **Findings** - **Uneven source splits.** The runner calls `BoundedSource.split(estimatedSize / parallelism)` once and keeps whatever comes back. Many sources don't honour the desired size. A BigQuery export, for example, returns one split per exported file: on one job, 714 files for 3750 readers left about 80% of the readers idle. Where there were enough splits, their sizes still varied a lot. On a 45 TiB Avro read at parallelism 3500, the busiest reader produced 1.38x the average records. - **Decode/re-encode in the batch sorter.** Flink's sorter copies records with `TypeSerializer#copy(DataInputView, DataOutputView)` when it spills. Beam coders don't delimit their output, so `CoderTypeSerializer` does a full decode and re-encode, which was about half of the task CPU in a profiled job. Records sent to GroupByKey and Combine are also decoded on arrival and immediately re-encoded into the sorter, although only their key is needed. - **Metric updates from source readers.** `ReaderInvocationUtil` pushes Beam metrics to Flink after every `start()`/`advance()` call. In a profiled job that was 37.6% of the read stage's CPU, more than reading the records. - **Windowing machinery in the global window.** The final step of GroupByKey and Combine runs `WindowDoFnOperator` with `ReduceFnRunner`, which binds state, holds and timers for every key. That bookkeeping was 30-40% of the reduce phase, against about 6% for merging the accumulators. - **Random Reshuffle assignment.** Flink's `rebalance()` starts each sender's round-robin at a random channel. When each sender emits only a few elements (file names, SDF restrictions), some receivers get nothing while others get several times the average. **Planned changes** (all batch-only; streaming pipelines are unchanged) - [ ] Re-split and balance bounded source splits, with static split assignment by default in batch (shared by the DataSet and DataStream translations) - [ ] Length-prefix batch records so the sorter can copy them as bytes, and decode the values of grouping inputs lazily (`--lengthPrefixBatchRecords`, `--lazyBatchGroupingInputs`) - [ ] Forward source reader metrics to Flink at most once per second (pipeline result metrics are unaffected) - [ ] A dedicated sorted operator for GroupByKey/Combine in the global window with the default trigger (`--sortedBatchCombine`) - [ ] Deterministic round-robin start channels for `Reshuffle` The new options are enabled by default and can be turned off with `--<option>=false`. Static split assignment can be reverted to lazy with `--sourceStaticSplitThresholdMb=0`. ### Issue Priority Priority: 2 (default / most feature requests should be filed as P2) ### Issue Components - [x] Component: Java SDK - [x] Component: Runner: Flink -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
