You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "acrites (via GitHub)" <gi...@apache.org> on 2026/10/16 20:31:46 UTC
[PR] Adds bytes read reporting for SDF. [beam]
acrites opened a new pull request, #40639:
URL: https://github.com/apache/beam/pull/40639
SDF's SplittableProcessFn's report their backlog for autoscaling, so we need to also report their bytes processed for autoscaling to correctly compute time to clear backlog.
SplittableProcessFn's are tagged when constructing MapTaskExecutors and their output coders are attached to the ParDoOperation so the counter values can be updated while processing.
Also adds tests for the new code.
--
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] Adds bytes read reporting for SDF. [beam]
Posted by "acrites (via GitHub)" <gi...@apache.org>.
acrites commented on code in PR #40639:
URL: https://github.com/apache/beam/pull/40639#discussion_r4404412315
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/common/worker/ParDoFn.java:
##########
@@ -37,4 +38,8 @@ public interface ParDoFn {
void finishBundle() throws Exception;
void abort() throws Exception;
+
+ default @Nullable DoFnInfo<?, ?> getDoFnInfo() throws Exception {
+ return null;
Review Comment:
There are 11 other production and 3 test classes that implement ParDoFn but don't actually wrap a DoFn:
* AssignWindowsParDoFn
* CreateIsmShardKeyAndSortKeyParDoFn
* GroupAlsoByWindowsParDoFn
* PairWithConstantKeyParDoFn
* BatchSideInputPGBKParDoFn
* StreamingSideInputPGBKParDoFn
* ReifyTimestampAndWindowsParDoFn
* SimplePartialGroupByKeyParDoFn
* StreamingPCollectionViewWriterParDoFn
* ToIsmRecordForMultimapParDoFn
* ValuesParDoFn
* NoopParDoFn in IntrinsicMapTaskExecutorTest.java and MapTaskExecutorTest.java
* TestParDoFn in ParDoOperationTest.java
There is one more class, BatchModeUngroupingParDoFn, that wraps ParDoFn, so we should probably have it's `getDoFnInfo` delegate to the underlying ParDoFn.
For all the others though, I think we'd just have to return null. I'll go ahead and remove the default method since I think it's nicer to at least force new classes to think about what to return.
--
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] Adds bytes read reporting for SDF. [beam]
Posted by "acrites (via GitHub)" <gi...@apache.org>.
acrites commented on code in PR #40639:
URL: https://github.com/apache/beam/pull/40639#discussion_r4327812520
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/ComputationWorkExecutorFactory.java:
##########
@@ -166,23 +168,27 @@ private static void trackAutoscalingBytesRead(
MapTask mapTask,
Nodes.ParallelInstructionNode readNode,
Coder<?> readCoder,
- ReadOperation readOperation,
+ Operation operation,
MapTaskExecutor mapTaskExecutor,
String counterName) {
- NameContext nameContext =
- NameContext.create(
- mapTask.getStageName(),
- readNode.getParallelInstruction().getOriginalName(),
- readNode.getParallelInstruction().getSystemName(),
- readNode.getParallelInstruction().getName());
- readOperation.receivers[0].addOutputCounter(
- counterName,
- new OutputObjectAndByteCounter(
- new IntrinsicMapTaskExecutorFactory.ElementByteSizeObservableCoder<>(readCoder),
- mapTaskExecutor.getOutputCounters(),
- nameContext)
- .setSamplingPeriod(100)
- .countBytes(counterName));
+ if (operation.receivers != null
+ && operation.receivers.length > 0
+ && operation.receivers[0] != null) {
+ NameContext nameContext =
+ NameContext.create(
+ mapTask.getStageName(),
+ readNode.getParallelInstruction().getOriginalName(),
+ readNode.getParallelInstruction().getSystemName(),
+ readNode.getParallelInstruction().getName());
+ operation.receivers[0].addOutputCounter(
Review Comment:
I'm not entirely sure. I didn't actually change this code, just added some extra null checks. I guess as long as the SplittableFn is not outputting the same element across multiple outputs we don't risk double-counting. I think that might more closely match what the portable worker does.
--
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] Adds bytes read reporting for SDF. [beam]
Posted by "acrites (via GitHub)" <gi...@apache.org>.
acrites commented on code in PR #40639:
URL: https://github.com/apache/beam/pull/40639#discussion_r4327857012
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/IntrinsicMapTaskExecutorFactory.java:
##########
@@ -272,18 +270,29 @@ private OperationNode createParDoOperation(
}
successorOffset += 1;
}
+ CloudObject cloudUserFn = CloudObject.fromSpec(parDo.getUserFn());
ParDoFn fn =
parDoFnFactory.create(
options,
- CloudObject.fromSpec(parDo.getUserFn()),
+ cloudUserFn,
parDo.getSideInputs(),
mainOutputTag,
outputTagsToReceiverIndicesBuilder.build(),
executionContext,
operationContext);
+ // An SDF's SplittableProcessFn behaves like a source, so we need to report its bytes read even
+ // though it's not a read operator for autoscaling to work properly.
+ Coder<?> bytesReadCoder = null;
+ if ("SplittableProcessFn".equals(cloudUserFn.getClassName())) {
Review Comment:
My first thought was to add something more generically to ParDo, but I feel that this is more of a Dataflow implementation detail that we even need to do this, so I'd rather not add something to the generic API to specify which ParDo's should be reporting their bytes to autoscaling. Maybe there's some way to add something to the CloudObject in SplittableProcessFnFactory.java?
--
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] Adds bytes read reporting for SDF. [beam]
Posted by "acrites (via GitHub)" <gi...@apache.org>.
acrites commented on code in PR #40639:
URL: https://github.com/apache/beam/pull/40639#discussion_r4394631240
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/IntrinsicMapTaskExecutorFactory.java:
##########
@@ -272,18 +270,29 @@ private OperationNode createParDoOperation(
}
successorOffset += 1;
}
+ CloudObject cloudUserFn = CloudObject.fromSpec(parDo.getUserFn());
ParDoFn fn =
parDoFnFactory.create(
options,
- CloudObject.fromSpec(parDo.getUserFn()),
+ cloudUserFn,
parDo.getSideInputs(),
mainOutputTag,
outputTagsToReceiverIndicesBuilder.build(),
executionContext,
operationContext);
+ // An SDF's SplittableProcessFn behaves like a source, so we need to report its bytes read even
+ // though it's not a read operator for autoscaling to work properly.
+ Coder<?> bytesReadCoder = null;
+ if ("SplittableProcessFn".equals(cloudUserFn.getClassName())) {
Review Comment:
I think I came up with a solution. It involves getting the (already deserialized) `DoFnInfo` from `ParDoFn` and then checking its signature as you suggested.
--
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] Adds bytes read reporting for SDF. [beam]
Posted by "acrites (via GitHub)" <gi...@apache.org>.
acrites commented on code in PR #40639:
URL: https://github.com/apache/beam/pull/40639#discussion_r4394618101
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/common/worker/MapTaskExecutor.java:
##########
@@ -145,6 +145,21 @@ public ReadOperation getReadOperation() {
return (ReadOperation) readOperation;
}
+ public List<ParDoOperation> getReadReportingParDoOperations() {
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] Adds bytes read reporting for SDF. [beam]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #40639:
URL: https://github.com/apache/beam/pull/40639#issuecomment-6225356096
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/40639?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 `90.69767%` with `4 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 76.03%. Comparing base ([`76f6e6d`](https://app.codecov.io/gh/apache/beam/commit/76f6e6d7c0c1b2ce582c73d068ad9260cf15c5f3?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`293c6ac`](https://app.codecov.io/gh/apache/beam/commit/293c6ac960dfb0ab806c2ed8534fa1788c03d73a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 215 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/beam/pull/40639?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...dataflow/worker/StreamingModeExecutionContext.java](https://app.codecov.io/gh/apache/beam/pull/40639?src=pr&el=tree&filepath=runners%2Fgoogle-cloud-dataflow-java%2Fworker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbeam%2Frunners%2Fdataflow%2Fworker%2FStreamingModeExecutionContext.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cnVubmVycy9nb29nbGUtY2xvdWQtZGF0YWZsb3ctamF2YS93b3JrZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2JlYW0vcnVubmVycy9kYXRhZmxvdy93b3JrZXIvU3RyZWFtaW5nTW9kZUV4ZWN1dGlvbkNvbnRleHQuamF2YQ==) | 90.00% | [0 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/beam/pull/40639?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...taflow/worker/IntrinsicMapTaskExecutorFactory.java](https://app.codecov.io/gh/apache/beam/pull/40639?src=pr&el=tree&filepath=runners%2Fgoogle-cloud-dataflow-java%2Fworker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbeam%2Frunners%2Fdataflow%2Fworker%2FIntrinsicMapTaskExecutorFactory.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cnVubmVycy9nb29nbGUtY2xvdWQtZGF0YWZsb3ctamF2YS93b3JrZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2JlYW0vcnVubmVycy9kYXRhZmxvdy93b3JrZXIvSW50cmluc2ljTWFwVGFza0V4ZWN1dG9yRmFjdG9yeS5qYXZh) | 88.88% | [0 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/beam/pull/40639?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...low/worker/util/common/worker/MapTaskExecutor.java](https://app.codecov.io/gh/apache/beam/pull/40639?src=pr&el=tree&filepath=runners%2Fgoogle-cloud-dataflow-java%2Fworker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fbeam%2Frunners%2Fdataflow%2Fworker%2Futil%2Fcommon%2Fworker%2FMapTaskExecutor.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cnVubmVycy9nb29nbGUtY2xvdWQtZGF0YWZsb3ctamF2YS93b3JrZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2JlYW0vcnVubmVycy9kYXRhZmxvdy93b3JrZXIvdXRpbC9jb21tb24vd29ya2VyL01hcFRhc2tFeGVjdXRvci5qYXZh) | 88.88% | [0 Missing and 1 partial :warning: ](https://app.codecov.io/gh/apache/beam/pull/40639?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 #40639 +/- ##
=============================================
+ Coverage 56.16% 76.03% +19.86%
- Complexity 2288 2301 +13
=============================================
Files 1124 259 -865
Lines 178336 14490 -163846
Branches 1489 1496 +7
=============================================
- Hits 100170 11018 -89152
+ Misses 75645 2766 -72879
+ Partials 2521 706 -1815
```
| [Flag](https://app.codecov.io/gh/apache/beam/pull/40639/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [go](https://app.codecov.io/gh/apache/beam/pull/40639/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java](https://app.codecov.io/gh/apache/beam/pull/40639/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.03% <90.69%> (+0.25%)` | :arrow_up: |
| [python](https://app.codecov.io/gh/apache/beam/pull/40639/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
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/40639?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] Adds bytes read reporting for SDF. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #40639:
URL: https://github.com/apache/beam/pull/40639#discussion_r4398449549
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/common/worker/ParDoFn.java:
##########
@@ -37,4 +38,8 @@ public interface ParDoFn {
void finishBundle() throws Exception;
void abort() throws Exception;
+
+ default @Nullable DoFnInfo<?, ?> getDoFnInfo() throws Exception {
+ return null;
Review Comment:
perhaps it would be better to remove the default so that new classes (and perhaps other existing ones) return an appropriate dofninfo?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingKeyedWorkItemSideInputParDoFn.java:
##########
@@ -231,18 +226,13 @@ protected void onProcessWindowedValue(WindowedValue<KeyedWorkItem<K, InputT>> el
}
if (hasState) {
helpers.registerStateCleanup(
- (WindowingStrategy<?, W>) getDoFnInfo().getWindowingStrategy(), windowsProcessed);
+ (WindowingStrategy<?, W>) helpers.fnInfo.getWindowingStrategy(), windowsProcessed);
}
}
- /**
- * Returns the {@link DoFnInfo} currently being used by this {@link SimpleParDoFn}.
- *
- * <p>May be null if no element has been processed yet, or if the {@link SimpleParDoFn} has
- * finished.
- */
- @VisibleForTesting
- @Nullable DoFnInfo<?, ?> getDoFnInfo() {
- return helpers.fnInfo;
+ /** Returns the {@link DoFnInfo} used by this {@link StreamingKeyedWorkItemSideInputParDoFn}. */
Review Comment:
hmm, it appears these DoFnInfo are stateful in that they contain a DoFn which goes through the dofn lifecycle. previously we would return only the active dofninfo (or null if not processing). Now we always return one but it might not be an active one or one that is then activated in another thread.
We don't want callers of getDoFnInfo to call methods that expect to just be used from the active thread (like dofn lifecycle etc). It seems like it would be cleanest if all the immutable stuff within DoFnInfo was a different interface which was returned so that getDoFnInfo couldn't be misused.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/SimpleParDoFn.java:
##########
@@ -112,7 +111,7 @@ protected void onStartKey() {
// These elements are now processed. Register cleanup timers for all the unblocked
// windows.
helpers.registerStateCleanup(
- (WindowingStrategy<?, W>) getDoFnInfo().getWindowingStrategy(),
+ (WindowingStrategy<?, W>) helpers.fnInfo.getWindowingStrategy(),
Review Comment:
maybe we should rename helpers fnInfo to activeFnInfo or something indicating it is the fninfo on the active thread?
See comment below
--
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] Adds bytes read reporting for SDF. [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #40639:
URL: https://github.com/apache/beam/pull/40639#issuecomment-6225706682
Assigning reviewers:
R: @kennknowles added as fallback since no labels match configuration
Backup reviewer: damccorm (if the main reviewer is unresponsive, you can escalate to them by tagging their username)
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] Adds bytes read reporting for SDF. [beam]
Posted by "acrites (via GitHub)" <gi...@apache.org>.
acrites commented on code in PR #40639:
URL: https://github.com/apache/beam/pull/40639#discussion_r4327819573
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/ComputationWorkExecutorFactory.java:
##########
@@ -166,23 +168,27 @@ private static void trackAutoscalingBytesRead(
MapTask mapTask,
Nodes.ParallelInstructionNode readNode,
Coder<?> readCoder,
- ReadOperation readOperation,
+ Operation operation,
MapTaskExecutor mapTaskExecutor,
String counterName) {
- NameContext nameContext =
- NameContext.create(
- mapTask.getStageName(),
- readNode.getParallelInstruction().getOriginalName(),
- readNode.getParallelInstruction().getSystemName(),
- readNode.getParallelInstruction().getName());
- readOperation.receivers[0].addOutputCounter(
- counterName,
- new OutputObjectAndByteCounter(
- new IntrinsicMapTaskExecutorFactory.ElementByteSizeObservableCoder<>(readCoder),
- mapTaskExecutor.getOutputCounters(),
- nameContext)
- .setSamplingPeriod(100)
- .countBytes(counterName));
+ if (operation.receivers != null
+ && operation.receivers.length > 0
+ && operation.receivers[0] != null) {
+ NameContext nameContext =
+ NameContext.create(
+ mapTask.getStageName(),
+ readNode.getParallelInstruction().getOriginalName(),
+ readNode.getParallelInstruction().getSystemName(),
+ readNode.getParallelInstruction().getName());
+ operation.receivers[0].addOutputCounter(
+ counterName,
+ new OutputObjectAndByteCounter(
+ new IntrinsicMapTaskExecutorFactory.ElementByteSizeObservableCoder<>(readCoder),
+ mapTaskExecutor.getOutputCounters(),
+ nameContext)
+ .setSamplingPeriod(100)
Review Comment:
I'm happy to change it if we can agree on a better time. This is just what we currently do for custom source reporting already.
--
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] Adds bytes read reporting for SDF. [beam]
Posted by "acrites (via GitHub)" <gi...@apache.org>.
acrites commented on PR #40639:
URL: https://github.com/apache/beam/pull/40639#issuecomment-6480899111
The code wasn't handling multi-output SDF's properly since it assumed we would be using the same coder for each output. So I fixed that and added some new tests.
--
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] Adds bytes read reporting for SDF. [beam]
Posted by "acrites (via GitHub)" <gi...@apache.org>.
acrites commented on code in PR #40639:
URL: https://github.com/apache/beam/pull/40639#discussion_r4405423752
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingKeyedWorkItemSideInputParDoFn.java:
##########
@@ -231,18 +226,13 @@ protected void onProcessWindowedValue(WindowedValue<KeyedWorkItem<K, InputT>> el
}
if (hasState) {
helpers.registerStateCleanup(
- (WindowingStrategy<?, W>) getDoFnInfo().getWindowingStrategy(), windowsProcessed);
+ (WindowingStrategy<?, W>) helpers.fnInfo.getWindowingStrategy(), windowsProcessed);
}
}
- /**
- * Returns the {@link DoFnInfo} currently being used by this {@link SimpleParDoFn}.
- *
- * <p>May be null if no element has been processed yet, or if the {@link SimpleParDoFn} has
- * finished.
- */
- @VisibleForTesting
- @Nullable DoFnInfo<?, ?> getDoFnInfo() {
- return helpers.fnInfo;
+ /** Returns the {@link DoFnInfo} used by this {@link StreamingKeyedWorkItemSideInputParDoFn}. */
Review Comment:
Yeah, I think I see what you're saying. One option would be to have separate methods for *all* of the things on the underlying DoFn that are immutable, e.g. WindowingStrategy. But that seems like a lot of extra boilerplate. Also, for the DoFnSignature stuff, I need an instance of the DoFn object itself.
So maybe we go with a combination of what you've suggested: change `helpers.fnInfo` to `helpers.activeFnInfo` to make it more clear that this is a dynamic object. And then have `getDoFnInfo` and `getActiveDoFnInfo` with better comments. `getActiveDoFnInfo` doesn't need to be part of the generic `ParDoFn` 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
Re: [PR] Adds bytes read reporting for SDF. [beam]
Posted by "acrites (via GitHub)" <gi...@apache.org>.
acrites commented on code in PR #40639:
URL: https://github.com/apache/beam/pull/40639#discussion_r4394654888
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/SimpleParDoFn.java:
##########
@@ -112,7 +111,7 @@ protected void onStartKey() {
// These elements are now processed. Register cleanup timers for all the unblocked
// windows.
helpers.registerStateCleanup(
- (WindowingStrategy<?, W>) getDoFnInfo().getWindowingStrategy(),
+ (WindowingStrategy<?, W>) helpers.fnInfo.getWindowingStrategy(),
Review Comment:
Note I changed these to call `helpers.fnInfo` directly since `getDoFnInfo()` now potentially throws `Exception` (from `helpers.doFnInstanceManager.peek()`). There were already instances of calling `helpers.fnInfo` in this method so this seemed cleaner than adding more exception throwing.
--
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] Adds bytes read reporting for SDF. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #40639:
URL: https://github.com/apache/beam/pull/40639#discussion_r4377728877
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/IntrinsicMapTaskExecutorFactory.java:
##########
@@ -272,18 +270,29 @@ private OperationNode createParDoOperation(
}
successorOffset += 1;
}
+ CloudObject cloudUserFn = CloudObject.fromSpec(parDo.getUserFn());
ParDoFn fn =
parDoFnFactory.create(
options,
- CloudObject.fromSpec(parDo.getUserFn()),
+ cloudUserFn,
parDo.getSideInputs(),
mainOutputTag,
outputTagsToReceiverIndicesBuilder.build(),
executionContext,
operationContext);
+ // An SDF's SplittableProcessFn behaves like a source, so we need to report its bytes read even
+ // though it's not a read operator for autoscaling to work properly.
+ Coder<?> bytesReadCoder = null;
+ if ("SplittableProcessFn".equals(cloudUserFn.getClassName())) {
Review Comment:
Is there a way to get the DoFnSignature and examine it's processElement.isSplittable()?
That seems to be done elsewhere to determine if something is a splittable dofn.
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/util/common/worker/MapTaskExecutor.java:
##########
@@ -145,6 +145,21 @@ public ReadOperation getReadOperation() {
return (ReadOperation) readOperation;
}
+ public List<ParDoOperation> getReadReportingParDoOperations() {
Review Comment:
would it be better to have a more generic method to get all the pardo operations or all operations and do the specific filtering to the bytesread ones done at the caller? seems like that would keep down on method proliferation
--
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] Adds bytes read reporting for SDF. [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on PR #40639:
URL: https://github.com/apache/beam/pull/40639#issuecomment-6294734662
R: @scwhittle
--
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] Adds bytes read reporting for SDF. [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #40639:
URL: https://github.com/apache/beam/pull/40639#issuecomment-6294756930
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment `assign set of reviewers`
--
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] Adds bytes read reporting for SDF. [beam]
Posted by "acrites (via GitHub)" <gi...@apache.org>.
acrites commented on code in PR #40639:
URL: https://github.com/apache/beam/pull/40639#discussion_r4342989905
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/IntrinsicMapTaskExecutorFactory.java:
##########
@@ -272,18 +270,29 @@ private OperationNode createParDoOperation(
}
successorOffset += 1;
}
+ CloudObject cloudUserFn = CloudObject.fromSpec(parDo.getUserFn());
ParDoFn fn =
parDoFnFactory.create(
options,
- CloudObject.fromSpec(parDo.getUserFn()),
+ cloudUserFn,
parDo.getSideInputs(),
mainOutputTag,
outputTagsToReceiverIndicesBuilder.build(),
executionContext,
operationContext);
+ // An SDF's SplittableProcessFn behaves like a source, so we need to report its bytes read even
+ // though it's not a read operator for autoscaling to work properly.
+ Coder<?> bytesReadCoder = null;
+ if ("SplittableProcessFn".equals(cloudUserFn.getClassName())) {
Review Comment:
Here were some alternatives provided by AI, though it's first recommendation is what I didn't really want to do:
1. Add a capability method to ParDoFn (Recommended)
Add default boolean isSplittable() { return false; } (or shouldReportBytesRead()) on ParDoFn (and delegate in ForwardingParDoFn).
Determine it via the actual DoFn class rather than any string: SplittableProcessFnFactory.ProcessFnExtractor wraps the user DoFn in SplittableParDoViaKeyedWorkItems.ProcessFn, which is already accessible at construction time via helpers.doFnInstanceManager.peek().getDoFn() instanceof ProcessFn inside SimpleParDoFnHelpers (used by both SimpleParDoFn and StreamingKeyedWorkItemSideInputParDoFn).
Alternatively, SplittableProcessFnFactory.createDefault() (or UserParDoFnFactory.streamingKeyedWorkItem, which is already true exclusively for SplittableProcessFnFactory) can set/wrap that flag on the returned ParDoFn.
IntrinsicMapTaskExecutorFactory then simply checks if (fn.isSplittable()), keeping all CloudObject dispatch encapsulated inside DefaultParDoFnFactory.
2. Have SplittableProcessFnFactory wrap with a dedicated ForwardingParDoFn subclass (or marker interface)
If you don't want to touch the public ParDoFn interface, SplittableProcessFnFactory can wrap the ParDoFn produced by UserParDoFnFactory in a package-private SplittableParDoFn extends ForwardingParDoFn (or marker interface), allowing fn instanceof SplittableParDoFn in IntrinsicMapTaskExecutorFactory.
3. Share a constant if keeping the CloudObject check
If you prefer keeping the check on cloudUserFn.getClassName(), extracting a shared constant (e.g., SplittableProcessFnFactory.OBJECT_TYPE_NAME = "SplittableProcessFn") used by DefaultParDoFnFactory, IntrinsicMapTaskExecutorFactory, and tests avoids duplicating the string literal (similar to ComputationWorkExecutorFactory.isCustomSource), though Option 1 or 2 is cleaner since it delegates the classification to ParDoFnFactory.
--
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] Adds bytes read reporting for SDF. [beam]
Posted by "scwhittle (via GitHub)" <gi...@apache.org>.
scwhittle commented on code in PR #40639:
URL: https://github.com/apache/beam/pull/40639#discussion_r4323275818
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/ComputationWorkExecutorFactory.java:
##########
@@ -166,23 +168,27 @@ private static void trackAutoscalingBytesRead(
MapTask mapTask,
Nodes.ParallelInstructionNode readNode,
Coder<?> readCoder,
- ReadOperation readOperation,
+ Operation operation,
MapTaskExecutor mapTaskExecutor,
String counterName) {
- NameContext nameContext =
- NameContext.create(
- mapTask.getStageName(),
- readNode.getParallelInstruction().getOriginalName(),
- readNode.getParallelInstruction().getSystemName(),
- readNode.getParallelInstruction().getName());
- readOperation.receivers[0].addOutputCounter(
- counterName,
- new OutputObjectAndByteCounter(
- new IntrinsicMapTaskExecutorFactory.ElementByteSizeObservableCoder<>(readCoder),
- mapTaskExecutor.getOutputCounters(),
- nameContext)
- .setSamplingPeriod(100)
- .countBytes(counterName));
+ if (operation.receivers != null
+ && operation.receivers.length > 0
+ && operation.receivers[0] != null) {
+ NameContext nameContext =
+ NameContext.create(
+ mapTask.getStageName(),
+ readNode.getParallelInstruction().getOriginalName(),
+ readNode.getParallelInstruction().getSystemName(),
+ readNode.getParallelInstruction().getName());
+ operation.receivers[0].addOutputCounter(
+ counterName,
+ new OutputObjectAndByteCounter(
+ new IntrinsicMapTaskExecutorFactory.ElementByteSizeObservableCoder<>(readCoder),
+ mapTaskExecutor.getOutputCounters(),
+ nameContext)
+ .setSamplingPeriod(100)
Review Comment:
do we want to sample this often? should it be time-based?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/windmill/work/processing/ComputationWorkExecutorFactory.java:
##########
@@ -166,23 +168,27 @@ private static void trackAutoscalingBytesRead(
MapTask mapTask,
Nodes.ParallelInstructionNode readNode,
Coder<?> readCoder,
- ReadOperation readOperation,
+ Operation operation,
MapTaskExecutor mapTaskExecutor,
String counterName) {
- NameContext nameContext =
- NameContext.create(
- mapTask.getStageName(),
- readNode.getParallelInstruction().getOriginalName(),
- readNode.getParallelInstruction().getSystemName(),
- readNode.getParallelInstruction().getName());
- readOperation.receivers[0].addOutputCounter(
- counterName,
- new OutputObjectAndByteCounter(
- new IntrinsicMapTaskExecutorFactory.ElementByteSizeObservableCoder<>(readCoder),
- mapTaskExecutor.getOutputCounters(),
- nameContext)
- .setSamplingPeriod(100)
- .countBytes(counterName));
+ if (operation.receivers != null
+ && operation.receivers.length > 0
+ && operation.receivers[0] != null) {
+ NameContext nameContext =
+ NameContext.create(
+ mapTask.getStageName(),
+ readNode.getParallelInstruction().getOriginalName(),
+ readNode.getParallelInstruction().getSystemName(),
+ readNode.getParallelInstruction().getName());
+ operation.receivers[0].addOutputCounter(
Review Comment:
why track just the first receiver? If a operator is multioutput are we missing tracking some of it?
##########
runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/IntrinsicMapTaskExecutorFactory.java:
##########
@@ -272,18 +270,29 @@ private OperationNode createParDoOperation(
}
successorOffset += 1;
}
+ CloudObject cloudUserFn = CloudObject.fromSpec(parDo.getUserFn());
ParDoFn fn =
parDoFnFactory.create(
options,
- CloudObject.fromSpec(parDo.getUserFn()),
+ cloudUserFn,
parDo.getSideInputs(),
mainOutputTag,
outputTagsToReceiverIndicesBuilder.build(),
executionContext,
operationContext);
+ // An SDF's SplittableProcessFn behaves like a source, so we need to report its bytes read even
+ // though it's not a read operator for autoscaling to work properly.
+ Coder<?> bytesReadCoder = null;
+ if ("SplittableProcessFn".equals(cloudUserFn.getClassName())) {
Review Comment:
this seems brittle. I'm not very familiar with this code but perhaps Kenn will have a cleaner way of doing this.
--
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