Fix DataflowOutputCounter calculation for ValueInEmptyWindows - #39487
Fix DataflowOutputCounter calculation for ValueInEmptyWindows#39487shunping wants to merge 3 commits into
Conversation
When processing shuffle or streaming data in Dataflow Legacy Runner (e.g., from GroupingShuffleReader or WindowingWindmillReader), KeyedWorkItems are wrapped inside a ValueInEmptyWindows (windows.size() == 0). Previously, DataflowOutputCounter.update() counted these as 1 element. This caused inaccurate element counts because: 1. A KeyedWorkItem can contain multiple elements. 2. Elements may belong to multiple windows and need to be fanned out accordingly. 3. KeyedWorkItems containing only timers were incorrectly incrementing element counters.
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
|
Caution The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased. |
| KeyedWorkItem<?, ?> keyedWorkItem = (KeyedWorkItem<?, ?>) value; | ||
| long totalElementCount = 0; | ||
| // Iterate only through elementsIterable and ignore timers in KeyedWorkItem. | ||
| for (WindowedValue<?> element : keyedWorkItem.elementsIterable()) { |
There was a problem hiding this comment.
I'm a little concerned this might hurt performance by adding another decoding of all the input
WindmillKeyedWorkItem is decoding as it iterates so the extra iteration is going to redecode everything. We could change WindmillKeyedWorkItem to decode once and cache, but that might have unintended effects as well since we are changing from keeping 1 element decoded in memory at time to all of them decoded at once.
This case and the extra iteration in ReduceFnRunner are just trying to observe the windows. We could optimize those by extending KeyedWorkItem interface to include a new
Iterable<WindowedValue<?>> elementWindowsIterable().
A default impl could just to retun elementsIterable().
But WindmillKeyedWorkItem could return a cheaper iterable that skips decoding the values, just the window metadata. And we could use the new method here and in ReduceFnRunner when we are collecting the windows.
Thoughts? @kennknowles if other ideas for interface
|
Assigning reviewers: R: @jrmccluskey 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). |
2 participants
When processing shuffle or streaming data in Dataflow Legacy Runner (e.g., from GroupingShuffleReader or WindowingWindmillReader), KeyedWorkItems are wrapped inside a ValueInEmptyWindows (windows.size() == 0).
Previously, DataflowOutputCounter.update() counted these as 1 element. This caused inaccurate element counts because:
Internal bug id: 509652238