Repository navigation
Conversation
…hannels Flink's rebalance() starts each sender's round-robin at a random channel. When senders emit only a few elements each, as when a Reshuffle redistributes file names or splittable DoFn restrictions, the resulting assignment is close to random: some receivers get no elements while others get several times the average. In batch, translate Reshuffle with DeterministicRebalance instead: sender i of M sends its j-th element to channel (floor(i * N / M) + j) mod N for N receivers, so already balanced inputs stay balanced. Elements are tagged with their sender's index and parallelism, since a Partitioner is not told which sender it runs in. Streaming pipelines keep using rebalance(). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #40612 +/- ##
============================================
+ Coverage 56.16% 65.16% +8.99%
- Complexity 2288 13994 +11706
============================================
Files 1124 1730 +606
Lines 178336 108497 -69839
Branches 1489 11307 +9818
============================================
- Hits 100170 70706 -29464
+ Misses 75645 33578 -42067
- Partials 2521 4213 +1692
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
Assigning reviewers: R: @shunping added as fallback since no labels match configuration Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
There was a problem hiding this comment.
📋 Review Summary
This PR replaces Flink's random-start DataStream#rebalance() with a deterministic round-robin rebalance (DeterministicRebalance) for batch Reshuffle transforms on the DataStream translator path, preventing Poisson-distributed load skew when senders emit only a few elements each. The partitioning logic and tests are well-designed and compatible across Flink 1.x and 2.0.
🔍 General Feedback
- The
CHANGES.mdentry was inadvertently placed inside the commented-out template block at the top of the file rather than under the active# [2.78.0] - Unreleasedsection. - Follow-up consideration:
FlinkStreamingPortablePipelineTranslator#translateReshuffle(in bothrunners/flink/src/...andrunners/flink/2.0/src/...) still unconditionally callsinputDataStream.rebalance(). Updating it when!context.isStreaming()would extend this improvement to portable batch pipelines running on DataStream (Flink 2.0 default and Flink 1.x with--useDataStreamForBatch=true).
💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.
|
|
||
| ## New Features / Improvements | ||
|
|
||
| * (Java) Flink runner: `Reshuffle` in batch distributes elements round-robin from deterministic start channels instead of random ones, so inputs with few elements per sender stay balanced ([#40607](https://github.com/apache/beam/issues/40607)). |
There was a problem hiding this comment.
🟡 Medium: This entry was added inside the HTML-commented <!-- Template --> block (# [2.XX.X] - Unreleased, lines 19–57) instead of the active # [2.78.0] - Unreleased -> ## New Features / Improvements section (around line 71). Please move it to the 2.78.0 section so it is included in the release notes.
|
Could you help with the review here? @pkuzmickas |
Addresses #40607.
Flink's
rebalance()starts each sender's round-robin at a random channel. When senders emit only a few elements each, as when aReshuffleredistributes file names or splittable DoFn restrictions, the resulting assignment is close to random: some receivers get no elements while others get several times the average.What changes
In batch,
Reshuffleis translated with a newDeterministicRebalance: senderiofMsends itsj-th element to channel(floor(i * N / M) + j) mod NforNreceivers, so already balanced inputs stay balanced. Elements are tagged with their sender's index and parallelism, since a FlinkPartitioneris not told which sender it runs in. Streaming pipelines keep usingrebalance().Testing
New
DeterministicRebalanceTest: balance with few elements per sender, a single sender, fewer and more senders than receivers, sender key round trips, and an end-to-end batchReshufflekeeping all elements.