pkuzmickas opened a new pull request, #41158:
URL: https://github.com/apache/beam/pull/41158
In batch on the DataStream translation path (always on the Flink 2 runner,
and with `--useDataStreamForBatch` on Flink 1), every `Combine.perKey` gets a
partial combine before the shuffle, implemented by
`PartialReduceBundleOperator` (#32440). It buffers each bundle in a Guava
`ArrayListMultimap` keyed by the raw key object and combines each multimap
entry when the bundle finishes.
Keys whose Java `equals`/`hashCode` are not consistent with their encoding
never compare equal. The common case is `byte[]` (including Scala
`Array[Byte]`). Every buffered element then forms its own group, and the
partial combine emits every input element unchanged as its own accumulator. The
final combine after the shuffle is still correct, because it groups by encoded
key, so results stay correct. But the pre-shuffle reduction is lost entirely,
and an accumulator is often larger than the input it wraps.
On a production batch job keyed by `byte[]`, the partial combine emitted as
many records as it received (about 486 billion), and the shuffled data grew
about 11× over its input instead of shrinking. The sort spill then ran out of
local disk. The same pipeline on the Flink 1 DataSet runner, which compares
encoded keys, cut that stage to about 2% of its input.
This change keys the buffer by `Coder#structuralValue` of the key, taken
from the operator's `KvCoder`. That matches how keys compare after the shuffle.
For coders that are consistent with `equals`, `structuralValue` returns the key
itself, so other keys are unaffected.
`HashingFlinkCombineRunner`/`SortingFlinkCombineRunner` already treat each
group as a single key (they take it from the first element), so the emitted key
is unchanged.
Changes:
- `PartialReduceBundleOperator`: buffer elements by
`keyCoder.structuralValue(key)`, both when processing and when restoring
buffered elements from operator state.
- New `PartialReduceBundleOperatorTest`:
- distinct `byte[]` instances with equal contents are combined (this fails
without the fix: all five inputs are emitted uncombined);
- `Long` keys are combined exactly as before.
Testing:
- `:runners:flink:2.0:test --tests '*PartialReduceBundleOperatorTest'` and
`:runners:flink:1.20:test --tests '*PartialReduceBundleOperatorTest'` pass.
Without the operator change, `combinesKeysThatAreEqualOnlyByEncoding` fails.
- `:runners:flink:2.0:spotlessCheck checkstyleMain checkstyleTest` pass.
- Full `:runners:flink:2.0:test` locally (JDK 17): 414 tests, 4 failures,
all unrelated:
- `FlinkSubmissionTest` (2×, `NoSuchFieldException: modifiers` on JDK 17,
also failing on master);
- `BeamFlinkDataStreamAdapterTest` (2×, failed only in the parallel full
run and passed in 3 of 3 isolated runs with this change).
------------------------
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).
--
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]
