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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
