pkuzmickas opened a new pull request, #41156: URL: https://github.com/apache/beam/pull/41156
Addresses #40607. Follow-up to #40612. `DeterministicRebalance` (#40612) starts sender `i` of `M` at channel `floor(i * N / M)`, so consecutive senders start on neighbouring channels. That keeps an input balanced when every sender has elements, but often only a block of consecutive subtasks has elements: a small `Create` or a few file patterns are assigned to the first subtasks, and only they emit the matched files. Once each of them emits more than `N / M` elements, their round-robin ranges overlap and pile onto the same few receivers. We hit this in a production batch pipeline: 48 file patterns on the first 48 of 1,760 subtasks matched 4,885 files. The files landed on about 200 readers, up to 48 each, while the rest got none. With `rebalance()` the busiest reader got about 6. One TaskManager ran out of local disk and the job failed. ## What changes Sender `i` now starts at channel `floor(s(i) * N / M)`, with `s(i) = i * g mod M`. Here `g` is the integer closest to `M` times the golden ratio conjugate that is coprime to `M`. - `s` is a permutation of the sender indices. The set of start channels is therefore unchanged, and so is the balance when every sender has elements; all existing tests pass unmodified. - Multiples of the golden ratio conjugate are spread almost evenly modulo one, so the start channels of any run of consecutive senders are spread over all receivers. Results with this partitioner (maximum elements per receiver, receivers with elements): | Senders with elements | Before | After | `rebalance()` (random starts) | |---|---|---|---| | 48 consecutive of 1,760, 4,885 elements | max 48, 148 of 1,760 receivers | max 4, all 1,760 | max 6 | | 200 consecutive of 2,000, 100 each | max 100, 299 of 2,000 receivers | max 11 (even: 10) | max 20 | | All senders (existing test cases) | balanced | identical | Poisson spread | ## Testing New tests in `DeterministicRebalanceTest`: - Consecutive busy senders at the start, middle and end spread over all receivers. - A block of busy senders stays close to balanced. - The start channels are a permutation of the evenly spaced channels for 1 to 600 senders and some large counts. The first two fail with the previous start channels (max 48 and max 100). All tests pass on `:runners:flink:2.0` and `:runners:flink:1.20`. `CHANGES.md`: amended the unreleased 2.79.0 entry for #40612. -- 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]
