jto opened a new pull request, #40608:
URL: https://github.com/apache/beam/pull/40608
Addresses #40607.
Improves how the Flink runner splits bounded sources in batch pipelines, so
that work is evenly distributed over the source readers. The same logic is
shared by the DataSet (`SourceInputFormat`) and DataStream (`FlinkSource`
enumerators) translations, in a new `BoundedSourceSplitter`.
## Motivation
The runner used to call `BoundedSource.split(estimatedSize / parallelism)`
once and use whatever came back. Beam does not require sources to split in one
pass, and many don't honour the desired size: for example a BigQuery export
returns one split per exported file, whatever size is requested. The result was
often too few splits for the readers, or splits of very uneven size. Combined
with lazy split assignment, a few readers could end up with much more work than
the others.
## What changes
After the initial `split()` call, the splitter refines the splits in a few
bounded passes:
1. **Re-split oversized splits.** Splits much larger than `estimatedSize /
(2 x parallelism)` are split again, so each reader gets at least two splits,
and sources returning splits far larger than requested (e.g. BigQuery exports)
are cut down.
2. **Balance sizes.** If split sizes still vary a lot (coefficient of
variation above 0.5), the larger splits are re-split towards the mean size, in
up to three rounds.
3. **Balance the split count.** If round-robin assignment would give the
busiest reader more than 10% above the average bytes, the largest splits are
halved.
4. **Order for round-robin.** Splits are sorted by size in a "snake" order,
so that assigning split `i` to reader `i % parallelism` pairs large splits with
small ones.
Steps 2 to 4 need per-split size estimates and are skipped when they are not
available. Re-splitting never goes below `min(64 MiB, estimatedSize / (2 x
parallelism))`: small splits mostly add fixed per-split overhead, but small
inputs, which may feed transforms that generate a lot more data, are still
spread over all readers. `fileInputSplitMaxSizeMB` is still honoured as a cap.
The DataStream enumerators now pass the actual reader count
(`currentParallelism()`) to the splitter, separately from `numSplits` (which
may be `maxParallelism`), so ordering and balancing match how splits are
actually assigned.
## Split assignment: static by default in batch
- `sourceStaticSplitThresholdMb` no longer has a default. When unset,
**batch pipelines use static assignment** and streaming pipelines keep lazy
assignment. Explicit values behave as before: `0` = lazy, negative = static,
positive = size-based.
- DataSet pipelines get a new `StaticInputSplitAssigner` (split `i` goes to
task `i % parallelism`, like the DataStream enumerator), and fall back to
Flink's `DefaultInputSplitAssigner` when lazy assignment is configured.
- The lazy enumerator hands out the largest splits first, in constant time
per assignment.
This is a behaviour change for batch pipelines that relied on the previous
lazy default. They can restore it with `--sourceStaticSplitThresholdMb=0`.
## Observability
Each step logs its effect at INFO, in human-readable units: split count,
size min/mean/max and coefficient of variation, and the busiest reader's load
relative to the average. For example:
Re-split 0 of 374596 splits larger than 192.00 MiB towards 128.00 MiB: ...
Split count balancing for 3500 readers: skipped, 374596 splits already
give the busiest reader 1.003 times the average bytes
Ordered 374596 splits for round-robin over 3500 readers: busiest reader
load / average went from 1.020 to 1.003
## Results
On a production-sized batch job (45 TiB of Avro, parallelism 3500,
DataStream translation), comparing the source stage before and after this
change, per subtask:
| | records out (max / mean) | records out CV | duration (max / mean) |
duration CV |
|---|---|---|---|---|
| before | 1.38 | 0.142 | 1.27 | 0.133 |
| after | 1.00 | 0.001 | 1.13 | 0.036 |
On a BigQuery export read (714 files for 3750 readers), the export files are
now re-split into 7854 splits, so all readers get work, with the busiest reader
at 1.017x the average bytes. Before, about 80% of the readers had nothing to
read.
## Trade-offs
- The refinement calls `getEstimatedSizeBytes()` on every split, and
`split()` again on some of them. For file-based sources this is in-memory work
once the file pattern has been expanded, but it may be slower for sources whose
size estimates are remote calls.
- Static assignment cannot adapt to readers that are slower for reasons
other than data volume (node, network, storage). Lazy assignment remains
available for that.
## Testing
- New `BoundedSourceSplitterTest`: re-splitting of oversized splits, size
balancing, split count balancing by bytes per reader, the minimum split size,
small inputs, round-robin ordering for the actual reader count, unsplittable
sources, assignment mode resolution and size formatting.
- `SourceInputFormatTest`: static assignment by default and lazy assignment
when requested.
- `FlinkSourceSplitEnumeratorTest` updated for the new default.
------------------------
Thank you for your contribution! Follow this checklist to help us
incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example:
`addresses #123`), if applicable. This will automatically add a link to the
pull request in the issue. If you would like the issue to automatically close
on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual
Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more
tips on [how to make review process
smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit
[https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more
information about GitHub Actions CI or the [workflows
README](https://github.com/apache/beam/blob/master/.github/workflows/README.md)
for a list of workflows and how to trigger them.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
