You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "shunping (via GitHub)" <gi...@apache.org> on 2026/07/25 03:02:49 UTC
[PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
shunping opened a new pull request, #39487:
URL: https://github.com/apache/beam/pull/39487
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.
Internal bug id: 509652238
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #39487:
URL: https://github.com/apache/beam/pull/39487#issuecomment-5121193537
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/39487?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `62.90323%` with `23 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 55.81%. Comparing base ([`dca1253`](https://app.codecov.io/gh/apache/beam/commit/dca1253c7e3886d125c526aa830b7945daef16ce?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`c1fb2d6`](https://app.codecov.io/gh/apache/beam/commit/c1fb2d6235fcc5c13548fe4acccfcc12be4484d0?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 100 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/beam/pull/39487?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...runners/dataflow/worker/WindmillKeyedWorkItem.java](https://app.codecov.io/gh/apache/beam/pull/39487?src=pr&el=tree&filepath=runners%2Fgoogle-cloud-dataflow-java%2Fworker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbeam%2Frunners%2Fdataflow%2Fworker%2FWindmillKeyedWorkItem.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cnVubmVycy9nb29nbGUtY2xvdWQtZGF0YWZsb3ctamF2YS93b3JrZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2JlYW0vcnVubmVycy9kYXRhZmxvdy93b3JrZXIvV2luZG1pbGxLZXllZFdvcmtJdGVtLmphdmE=) | 41.66% | [13 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/beam/pull/39487?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ataflow/worker/StreamingDataflowOutputCounter.java](https://app.codecov.io/gh/apache/beam/pull/39487?src=pr&el=tree&filepath=runners%2Fgoogle-cloud-dataflow-java%2Fworker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbeam%2Frunners%2Fdataflow%2Fworker%2FStreamingDataflowOutputCounter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cnVubmVycy9nb29nbGUtY2xvdWQtZGF0YWZsb3ctamF2YS93b3JrZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2JlYW0vcnVubmVycy9kYXRhZmxvdy93b3JrZXIvU3RyZWFtaW5nRGF0YWZsb3dPdXRwdXRDb3VudGVyLmphdmE=) | 68.75% | [3 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/beam/pull/39487?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...rs/dataflow/worker/BatchDataflowOutputCounter.java](https://app.codecov.io/gh/apache/beam/pull/39487?src=pr&el=tree&filepath=runners%2Fgoogle-cloud-dataflow-java%2Fworker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbeam%2Frunners%2Fdataflow%2Fworker%2FBatchDataflowOutputCounter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cnVubmVycy9nb29nbGUtY2xvdWQtZGF0YWZsb3ctamF2YS93b3JrZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2JlYW0vcnVubmVycy9kYXRhZmxvdy93b3JrZXIvQmF0Y2hEYXRhZmxvd091dHB1dENvdW50ZXIuamF2YQ==) | 66.66% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/beam/pull/39487?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...runners/dataflow/worker/DataflowOutputCounter.java](https://app.codecov.io/gh/apache/beam/pull/39487?src=pr&el=tree&filepath=runners%2Fgoogle-cloud-dataflow-java%2Fworker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbeam%2Frunners%2Fdataflow%2Fworker%2FDataflowOutputCounter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cnVubmVycy9nb29nbGUtY2xvdWQtZGF0YWZsb3ctamF2YS93b3JrZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2JlYW0vcnVubmVycy9kYXRhZmxvdy93b3JrZXIvRGF0YWZsb3dPdXRwdXRDb3VudGVyLmphdmE=) | 71.42% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/beam/pull/39487?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #39487 +/- ##
============================================
- Coverage 58.05% 55.81% -2.24%
+ Complexity 13057 2272 -10785
============================================
Files 2518 1115 -1403
Lines 263812 174219 -89593
Branches 10765 1460 -9305
============================================
- Hits 153164 97246 -55918
+ Misses 104899 74518 -30381
+ Partials 5749 2455 -3294
```
| [Flag](https://app.codecov.io/gh/apache/beam/pull/39487/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [java](https://app.codecov.io/gh/apache/beam/pull/39487/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `75.54% <62.90%> (+11.31%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/beam/pull/39487?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #39487:
URL: https://github.com/apache/beam/pull/39487#discussion_r3680046768
##########
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:
confusing comment, this may be a KeyedWorkItem
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping merged PR #39487:
URL: https://github.com/apache/beam/pull/39487
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #39487:
URL: https://github.com/apache/beam/pull/39487#discussion_r3677579253
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowOutputCounter.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker;
+
+import org.apache.beam.runners.core.ElementByteSizeObservable;
+import org.apache.beam.runners.core.KeyedWorkItem;
+import org.apache.beam.runners.dataflow.worker.counters.CounterFactory;
+import org.apache.beam.runners.dataflow.worker.counters.NameContext;
+import org.apache.beam.sdk.values.WindowedValue;
+
+/**
+ * A Dataflow output counter specific to Streaming pipelines. Unpacks {@link KeyedWorkItem}s in
+ * empty windows (e.g. emitted by WindowingWindmillReader) and counts element windows (ignoring
+ * timers) using lightweight metadata-only iteration via {@link
+ * KeyedWorkItem#elementWindowsIterable()}.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+public class StreamingDataflowOutputCounter extends DataflowOutputCounter {
+
+ public StreamingDataflowOutputCounter(
+ String outputName, CounterFactory counterFactory, NameContext nameContext) {
+ super(outputName, counterFactory, nameContext);
+ }
+
+ public StreamingDataflowOutputCounter(
+ String outputName,
+ ElementByteSizeObservable<?> elementByteSizeObservable,
+ CounterFactory counterFactory,
+ NameContext nameContext) {
+ super(outputName, elementByteSizeObservable, counterFactory, nameContext);
+ }
+
+ @Override
+ protected void updateEmptyWindows(Object elem) {
Review Comment:
Ok. Let's remove these subclasses then.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #39487:
URL: https://github.com/apache/beam/pull/39487#discussion_r3677734070
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowOutputCounter.java:
##########
@@ -41,20 +42,57 @@ public class DataflowOutputCounter implements ElementCounter {
private OutputObjectAndByteCounter objectAndByteCounter;
private Counter<Long, ?> elementCount;
+ private final boolean isStreaming;
+
+ public static DataflowOutputCounter create(
+ String outputName,
+ ElementByteSizeObservable<?> elementByteSizeObservable,
+ CounterFactory counterFactory,
+ NameContext nameContext,
+ boolean isStreaming) {
+ return new DataflowOutputCounter(
+ outputName, elementByteSizeObservable, counterFactory, nameContext, isStreaming);
+ }
+
+ public static DataflowOutputCounter create(
+ String outputName,
+ CounterFactory counterFactory,
+ NameContext nameContext,
+ boolean isStreaming) {
+ return create(outputName, null, counterFactory, nameContext, isStreaming);
+ }
public DataflowOutputCounter(
String outputName, CounterFactory counterFactory, NameContext nameContext) {
- this(outputName, null, counterFactory, nameContext);
+ this(outputName, null, counterFactory, nameContext, false);
+ }
+
+ public DataflowOutputCounter(
+ String outputName,
+ CounterFactory counterFactory,
+ NameContext nameContext,
+ boolean isStreaming) {
+ this(outputName, null, counterFactory, nameContext, isStreaming);
}
public DataflowOutputCounter(
String outputName,
ElementByteSizeObservable<?> elementByteSizeObservable,
CounterFactory counterFactory,
NameContext nameContext) {
- objectAndByteCounter =
+ this(outputName, elementByteSizeObservable, counterFactory, nameContext, false);
+ }
+
+ public DataflowOutputCounter(
Review Comment:
can we consolidate to just a single private constructor and just have the multiple create factories?
I think this class could be marked with Internal annotation as well
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowOutputCounter.java:
##########
@@ -63,15 +101,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);
}
}
+ protected void updateEmptyWindows(WindowedValue<?> elem) {
Review Comment:
think this could be private now
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/IntrinsicMapTaskExecutorFactory.java:
##########
@@ -346,6 +348,11 @@ OperationNode createFlattenOperation(
*/
static Function<Node, Node> createOutputReceiversTransform(
Review Comment:
can this be removed and just have the 3 param version? since it's internal to runner?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #39487:
URL: https://github.com/apache/beam/pull/39487#issuecomment-5094968358
Assigning reviewers:
R: @jrmccluskey added as fallback since no labels match configuration
Note: If you would like to opt out of this review, comment `assign to next reviewer`.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #39487:
URL: https://github.com/apache/beam/pull/39487#discussion_r3677575596
##########
runners/core-java/src/main/java/org/apache/beam/runners/core/KeyedWorkItem.java:
##########
@@ -35,4 +35,13 @@ public interface KeyedWorkItem<K, ElemT> {
/** Returns an iterable containing the elements. */
Iterable<WindowedValue<ElemT>> elementsIterable();
+
+ /**
+ * Returns an iterable containing windowed values without guaranteeing element payload decoding.
+ * Useful for lightweight inspection of windowing metadata without payload deserialization
+ * overhead.
+ */
+ default Iterable<WindowedValue<ElemT>> elementWindowsIterable() {
Review Comment:
Done.
##########
runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java:
##########
@@ -361,13 +361,24 @@ private Collection<W> windowsThatShouldFire(Set<W> windows) throws Exception {
* setting holds, and invoking {@link ReduceFn#onTrigger}.
* </ol>
*/
+ public void processElements(KeyedWorkItem<?, InputT> keyedWorkItem) throws Exception {
+ processElementsInternal(
+ keyedWorkItem.elementWindowsIterable(), keyedWorkItem.elementsIterable());
+ }
+
public void processElements(Iterable<WindowedValue<InputT>> values) throws Exception {
- if (!values.iterator().hasNext()) {
+ processElementsInternal(values, values);
+ }
+
+ private void processElementsInternal(
+ Iterable<WindowedValue<InputT>> elementWindows, Iterable<WindowedValue<InputT>> values)
+ throws Exception {
+ if (!elementWindows.iterator().hasNext()) {
Review Comment:
Done.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #39487:
URL: https://github.com/apache/beam/pull/39487#discussion_r3679068286
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowOutputCounter.java:
##########
@@ -63,15 +101,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);
}
}
+ protected void updateEmptyWindows(WindowedValue<?> elem) {
Review Comment:
Done
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowOutputCounter.java:
##########
@@ -41,20 +42,57 @@ public class DataflowOutputCounter implements ElementCounter {
private OutputObjectAndByteCounter objectAndByteCounter;
private Counter<Long, ?> elementCount;
+ private final boolean isStreaming;
+
+ public static DataflowOutputCounter create(
+ String outputName,
+ ElementByteSizeObservable<?> elementByteSizeObservable,
+ CounterFactory counterFactory,
+ NameContext nameContext,
+ boolean isStreaming) {
+ return new DataflowOutputCounter(
+ outputName, elementByteSizeObservable, counterFactory, nameContext, isStreaming);
+ }
+
+ public static DataflowOutputCounter create(
+ String outputName,
+ CounterFactory counterFactory,
+ NameContext nameContext,
+ boolean isStreaming) {
+ return create(outputName, null, counterFactory, nameContext, isStreaming);
+ }
public DataflowOutputCounter(
String outputName, CounterFactory counterFactory, NameContext nameContext) {
- this(outputName, null, counterFactory, nameContext);
+ this(outputName, null, counterFactory, nameContext, false);
+ }
+
+ public DataflowOutputCounter(
+ String outputName,
+ CounterFactory counterFactory,
+ NameContext nameContext,
+ boolean isStreaming) {
+ this(outputName, null, counterFactory, nameContext, isStreaming);
}
public DataflowOutputCounter(
String outputName,
ElementByteSizeObservable<?> elementByteSizeObservable,
CounterFactory counterFactory,
NameContext nameContext) {
- objectAndByteCounter =
+ this(outputName, elementByteSizeObservable, counterFactory, nameContext, false);
+ }
+
+ public DataflowOutputCounter(
Review Comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #39487:
URL: https://github.com/apache/beam/pull/39487#discussion_r3677571118
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/BatchDataflowOutputCounter.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker;
+
+import org.apache.beam.runners.core.ElementByteSizeObservable;
+import org.apache.beam.runners.dataflow.worker.counters.CounterFactory;
+import org.apache.beam.runners.dataflow.worker.counters.NameContext;
+
+/**
+ * A Dataflow output counter specific to Batch pipelines. In batch pipelines, empty-window elements
+ * (e.g. GroupingShuffleReader emitting KV<K, ValuesIterable>) represent a single PCollection
+ * element output.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
The file has been removed.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowOutputCounter.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker;
+
+import org.apache.beam.runners.core.ElementByteSizeObservable;
+import org.apache.beam.runners.core.KeyedWorkItem;
+import org.apache.beam.runners.dataflow.worker.counters.CounterFactory;
+import org.apache.beam.runners.dataflow.worker.counters.NameContext;
+import org.apache.beam.sdk.values.WindowedValue;
+
+/**
+ * A Dataflow output counter specific to Streaming pipelines. Unpacks {@link KeyedWorkItem}s in
+ * empty windows (e.g. emitted by WindowingWindmillReader) and counts element windows (ignoring
+ * timers) using lightweight metadata-only iteration via {@link
+ * KeyedWorkItem#elementWindowsIterable()}.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
The file has been removed.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #39487:
URL: https://github.com/apache/beam/pull/39487#discussion_r3676393013
##########
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:
Done. Added `elementWindowsIterable` to decode window metadata only. PTAL
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on PR #39487:
URL: https://github.com/apache/beam/pull/39487#issuecomment-5104486491
Based on the description of the problem, would it make sense to find a way to factor the code so we can have separate streaming/batch implementation of the output counter? The ways they pass elements is just pretty fundamentally different.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on PR #39487:
URL: https://github.com/apache/beam/pull/39487#issuecomment-5120979525
> Based on the description of the problem, would it make sense to find a way to factor the code so we can have separate streaming/batch implementation of the output counter? The ways they pass elements is just pretty fundamentally different.
The difference between the streaming and batch output counters really only matters during the GBK step, specifically regarding what gets wrapped inside `ValueInEmptyWindows`. Per your comment, I updated the PR to separate the implementations. Particularly I subclassed `DataflowOutputCounter` and extracted the specific routine that handles the `window_size == 0` case.
PTAL! Thanks!
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #39487:
URL: https://github.com/apache/beam/pull/39487#issuecomment-5076807431
> [!CAUTION]
> The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
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 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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #39487:
URL: https://github.com/apache/beam/pull/39487#discussion_r3677573581
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowOutputCounter.java:
##########
@@ -63,15 +84,17 @@ 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(elem);
} else {
+ // Standard WindowedValue.
elementCount.addValue(windowsSize);
}
}
+ protected void updateEmptyWindows(Object elem) throws Exception {
Review Comment:
Done.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillKeyedWorkItem.java:
##########
@@ -184,6 +184,56 @@ public Iterable<TimerData> timersIterable() {
}
}
+ @SuppressWarnings("nullness")
+ private @Nullable WindowedValue<ElemT> parseElemWindowOnly(Windmill.Message message) {
Review Comment:
Done.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "shunping (via GitHub)" <gi...@apache.org>.
shunping commented on code in PR #39487:
URL: https://github.com/apache/beam/pull/39487#discussion_r3679069022
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/IntrinsicMapTaskExecutorFactory.java:
##########
@@ -346,6 +348,11 @@ OperationNode createFlattenOperation(
*/
static Function<Node, Node> createOutputReceiversTransform(
Review Comment:
Done
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "gemini-code-assist[bot] (via GitHub)" <gi...@apache.org>.
gemini-code-assist[bot] commented on PR #39487:
URL: https://github.com/apache/beam/pull/39487#issuecomment-5076578648
> [!CAUTION]
> The consumer version of Gemini Code Assist on GitHub has been sunset. All code review activity has officially ceased.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #39487:
URL: https://github.com/apache/beam/pull/39487#discussion_r3676770885
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/BatchDataflowOutputCounter.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker;
+
+import org.apache.beam.runners.core.ElementByteSizeObservable;
+import org.apache.beam.runners.dataflow.worker.counters.CounterFactory;
+import org.apache.beam.runners.dataflow.worker.counters.NameContext;
+
+/**
+ * A Dataflow output counter specific to Batch pipelines. In batch pipelines, empty-window elements
+ * (e.g. GroupingShuffleReader emitting KV<K, ValuesIterable>) represent a single PCollection
+ * element output.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
can this be removed?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/DataflowOutputCounter.java:
##########
@@ -63,15 +84,17 @@ 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(elem);
} else {
+ // Standard WindowedValue.
elementCount.addValue(windowsSize);
}
}
+ protected void updateEmptyWindows(Object elem) throws Exception {
Review Comment:
have this take a WindowedValue<?> instead of Object
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillKeyedWorkItem.java:
##########
@@ -184,6 +184,56 @@ public Iterable<TimerData> timersIterable() {
}
}
+ @SuppressWarnings("nullness")
+ private @Nullable WindowedValue<ElemT> parseElemWindowOnly(Windmill.Message message) {
Review Comment:
can this share impl with parseElem? could delegate to internal method taking a bool on if the value should be parsed. Otherwise it might be easy to become inconsistent between the two
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowOutputCounter.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker;
+
+import org.apache.beam.runners.core.ElementByteSizeObservable;
+import org.apache.beam.runners.core.KeyedWorkItem;
+import org.apache.beam.runners.dataflow.worker.counters.CounterFactory;
+import org.apache.beam.runners.dataflow.worker.counters.NameContext;
+import org.apache.beam.sdk.values.WindowedValue;
+
+/**
+ * A Dataflow output counter specific to Streaming pipelines. Unpacks {@link KeyedWorkItem}s in
+ * empty windows (e.g. emitted by WindowingWindmillReader) and counts element windows (ignoring
+ * timers) using lightweight metadata-only iteration via {@link
+ * KeyedWorkItem#elementWindowsIterable()}.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
Review Comment:
ditto can this be removed?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowOutputCounter.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.runners.dataflow.worker;
+
+import org.apache.beam.runners.core.ElementByteSizeObservable;
+import org.apache.beam.runners.core.KeyedWorkItem;
+import org.apache.beam.runners.dataflow.worker.counters.CounterFactory;
+import org.apache.beam.runners.dataflow.worker.counters.NameContext;
+import org.apache.beam.sdk.values.WindowedValue;
+
+/**
+ * A Dataflow output counter specific to Streaming pipelines. Unpacks {@link KeyedWorkItem}s in
+ * empty windows (e.g. emitted by WindowingWindmillReader) and counts element windows (ignoring
+ * timers) using lightweight metadata-only iteration via {@link
+ * KeyedWorkItem#elementWindowsIterable()}.
+ */
+@SuppressWarnings({
+ "nullness" // TODO(https://github.com/apache/beam/issues/20497)
+})
+public class StreamingDataflowOutputCounter extends DataflowOutputCounter {
+
+ public StreamingDataflowOutputCounter(
+ String outputName, CounterFactory counterFactory, NameContext nameContext) {
+ super(outputName, counterFactory, nameContext);
+ }
+
+ public StreamingDataflowOutputCounter(
+ String outputName,
+ ElementByteSizeObservable<?> elementByteSizeObservable,
+ CounterFactory counterFactory,
+ NameContext nameContext) {
+ super(outputName, elementByteSizeObservable, counterFactory, nameContext);
+ }
+
+ @Override
+ protected void updateEmptyWindows(Object elem) {
Review Comment:
if this is the only difference between the two, I might just prefer removing the heirarchy and just having a bool member variable on whether to introspect KeyedWorkItem values
##########
runners/core-java/src/main/java/org/apache/beam/runners/core/KeyedWorkItem.java:
##########
@@ -35,4 +35,13 @@ public interface KeyedWorkItem<K, ElemT> {
/** Returns an iterable containing the elements. */
Iterable<WindowedValue<ElemT>> elementsIterable();
+
+ /**
+ * Returns an iterable containing windowed values without guaranteeing element payload decoding.
+ * Useful for lightweight inspection of windowing metadata without payload deserialization
+ * overhead.
+ */
+ default Iterable<WindowedValue<ElemT>> elementWindowsIterable() {
Review Comment:
can this return WindowedValue<?> or WindowedValue<Void> so that callers know not to use the value?
##########
runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java:
##########
@@ -361,13 +361,24 @@ private Collection<W> windowsThatShouldFire(Set<W> windows) throws Exception {
* setting holds, and invoking {@link ReduceFn#onTrigger}.
* </ol>
*/
+ public void processElements(KeyedWorkItem<?, InputT> keyedWorkItem) throws Exception {
+ processElementsInternal(
+ keyedWorkItem.elementWindowsIterable(), keyedWorkItem.elementsIterable());
+ }
+
public void processElements(Iterable<WindowedValue<InputT>> values) throws Exception {
- if (!values.iterator().hasNext()) {
+ processElementsInternal(values, values);
+ }
+
+ private void processElementsInternal(
+ Iterable<WindowedValue<InputT>> elementWindows, Iterable<WindowedValue<InputT>> values)
+ throws Exception {
+ if (!elementWindows.iterator().hasNext()) {
Review Comment:
nit: use Iterables.empty()
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix DataflowOutputCounter calculation for ValueInEmptyWindows [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org