scwhittle commented on code in PR #39487:
URL: https://github.com/apache/beam/pull/39487#discussion_r3659626752
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowOutputCounter.java:
##########
@@ -63,11 +64,27 @@ public void update(Object elem) throws Exception {
objectAndByteCounter.update(elem);
long windowsSize = ((WindowedValue<?>) elem).getWindows().size();
if (windowsSize == 0) {
- // GroupingShuffleReader produces ValueInEmptyWindows.
- // For now, we count the element at least once to keep the current
counter
- // behavior.
- elementCount.addValue(1L);
+ // ValueInEmptyWindows occurs when processing shuffle/streaming work
items
+ Object value = ((WindowedValue<?>) elem).getValue();
+ if (value instanceof KeyedWorkItem<?, ?>) {
+ // KeyedWorkItem wrapped in ValueInEmptyWindows
+ // (e.g. WindowingWindmillReader for Streaming GBK)
+ KeyedWorkItem<?, ?> keyedWorkItem = (KeyedWorkItem<?, ?>) value;
+ long totalElementCount = 0;
+ // Iterate only through elementsIterable and ignore timers in
KeyedWorkItem.
+ for (WindowedValue<?> element : keyedWorkItem.elementsIterable()) {
Review 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
--
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]
