shunping commented on code in PR #39487:
URL: https://github.com/apache/beam/pull/39487#discussion_r3682894650
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowOutputCounter.java:
##########
@@ -63,15 +82,42 @@ 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);
+ updateEmptyWindows((WindowedValue<?>) elem);
} else {
+ // Standard WindowedValue.
elementCount.addValue(windowsSize);
}
}
+ private void updateEmptyWindows(WindowedValue<?> elem) {
+ if (isStreaming) {
+ Object value = elem.getValue();
+ if (value instanceof KeyedWorkItem<?, ?>) {
+ // KeyedWorkItem wrapped in ValueInEmptyWindows
+ // (e.g. WindowingWindmillReader for Streaming GBK)
+ KeyedWorkItem<?, ?> keyedWorkItem = (KeyedWorkItem<?, ?>) value;
+ long totalElementCount = 0;
+ // Iterate through elementWindowsIterable and ignore timers in
KeyedWorkItem.
+ // Uses lightweight metadata-only iteration without payload
deserialization overhead.
+ for (WindowedValue<?> element :
keyedWorkItem.elementWindowsIterable()) {
+ long elementWindowsSize = element.getWindows().size();
+ // Fan out for windows.
+ totalElementCount += (elementWindowsSize == 0 ? 1L :
elementWindowsSize);
+ }
+ elementCount.addValue(totalElementCount);
+ } else {
+ // NOTE: in streaming mode, this should not normally happen.
+ // Counting as 1 element serves as a fallback to maintain counter
behavior without failing
+ // execution.
+ elementCount.addValue(1L);
+ }
+ } else {
+ // Non-KeyedWorkItem wrapped in ValueInEmptyWindows
Review Comment:
I think this is the branch for batch GBK, and we only receive KV (which is
not KeyedWorkWorkItem) wrapped in `ValueInEmptyWindows` from
`GroupingShuffleReader`.
https://github.com/apache/beam/blob/2fd6af992b58baa275da4f267d327a6f2cea26ca/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/GroupingShuffleReader.java#L293-L296
Did I miss any scenario?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowOutputCounter.java:
##########
@@ -63,15 +82,42 @@ 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);
+ updateEmptyWindows((WindowedValue<?>) elem);
} else {
+ // Standard WindowedValue.
elementCount.addValue(windowsSize);
}
}
+ private void updateEmptyWindows(WindowedValue<?> elem) {
+ if (isStreaming) {
+ Object value = elem.getValue();
+ if (value instanceof KeyedWorkItem<?, ?>) {
+ // KeyedWorkItem wrapped in ValueInEmptyWindows
+ // (e.g. WindowingWindmillReader for Streaming GBK)
+ KeyedWorkItem<?, ?> keyedWorkItem = (KeyedWorkItem<?, ?>) value;
+ long totalElementCount = 0;
+ // Iterate through elementWindowsIterable and ignore timers in
KeyedWorkItem.
+ // Uses lightweight metadata-only iteration without payload
deserialization overhead.
+ for (WindowedValue<?> element :
keyedWorkItem.elementWindowsIterable()) {
+ long elementWindowsSize = element.getWindows().size();
+ // Fan out for windows.
+ totalElementCount += (elementWindowsSize == 0 ? 1L :
elementWindowsSize);
+ }
+ elementCount.addValue(totalElementCount);
+ } else {
+ // NOTE: in streaming mode, this should not normally happen.
+ // Counting as 1 element serves as a fallback to maintain counter
behavior without failing
+ // execution.
+ elementCount.addValue(1L);
+ }
+ } else {
+ // Non-KeyedWorkItem wrapped in ValueInEmptyWindows
Review Comment:
I think this is the branch for batch GBK, and we only receive KV (which is
not `KeyedWorkWorkItem`) wrapped in `ValueInEmptyWindows` from
`GroupingShuffleReader`.
https://github.com/apache/beam/blob/2fd6af992b58baa275da4f267d327a6f2cea26ca/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/GroupingShuffleReader.java#L293-L296
Did I miss any scenario?
--
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]
