Skip to content

[BEAM-5041] Java Fn SDK Harness use pTransform to track processed graph #6093

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lukecwik merged 2 commits into from
Aug 8, 2018
Merged

[BEAM-5041] Java Fn SDK Harness use pTransform to track processed graph #6093

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8d96644
Use pTransforms to keep track of processed graph.
angoenka Jul 30, 2018
5b9fa2d
Minor method parameter type change.
lukecwik Aug 6, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 25 additions & 21 deletions sdks/java/harness/src/main/java/org/apache/beam/fn/harness/control/ProcessBundleHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import com.google.common.collect.Sets;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.ServiceLoader;
Expand Down Expand Up @@ -145,6 +146,7 @@ private void createRunnerAndConsumersForPTransformRecursively(
ProcessBundleDescriptor processBundleDescriptor,
SetMultimap<String, String> pCollectionIdsToConsumingPTransforms,
ListMultimap<String, FnDataReceiver<WindowedValue<?>>> pCollectionIdsToConsumers,
Set<String> processedPTransformIds,
Consumer<ThrowingRunnable> addStartFunction,
Consumer<ThrowingRunnable> addFinishFunction,
BundleSplitListener splitListener)
Expand All @@ -154,10 +156,6 @@ private void createRunnerAndConsumersForPTransformRecursively(
// Since we are creating the consumers first, we know that the we are building the DAG
// in reverse topological order.
for (String pCollectionId : pTransform.getOutputsMap().values()) {
// If we have created the consumers for this PCollection we can skip it.
if (pCollectionIdsToConsumers.containsKey(pCollectionId)) {
continue;
}

for (String consumingPTransformId : pCollectionIdsToConsumingPTransforms.get(pCollectionId)) {
createRunnerAndConsumersForPTransformRecursively(
Expand All @@ -168,6 +166,7 @@ private void createRunnerAndConsumersForPTransformRecursively(
processBundleDescriptor,
pCollectionIdsToConsumingPTransforms,
pCollectionIdsToConsumers,
processedPTransformIds,
addStartFunction,
addFinishFunction,
splitListener);
Expand All @@ -185,23 +184,26 @@ private void createRunnerAndConsumersForPTransformRecursively(
String.format(
"Cannot process composite transform: %s", TextFormat.printToString(pTransform)));
}

urnToPTransformRunnerFactoryMap
.getOrDefault(pTransform.getSpec().getUrn(), defaultPTransformRunnerFactory)
.createRunnerForPTransform(
options,
beamFnDataClient,
beamFnStateClient,
pTransformId,
pTransform,
processBundleInstructionId,
processBundleDescriptor.getPcollectionsMap(),
processBundleDescriptor.getCodersMap(),
processBundleDescriptor.getWindowingStrategiesMap(),
pCollectionIdsToConsumers,
addStartFunction,
addFinishFunction,
splitListener);
// Skip reprocessing processed pTransforms.
if (!processedPTransformIds.contains(pTransformId)) {
urnToPTransformRunnerFactoryMap
.getOrDefault(pTransform.getSpec().getUrn(), defaultPTransformRunnerFactory)
.createRunnerForPTransform(
options,
beamFnDataClient,
beamFnStateClient,
pTransformId,
pTransform,
processBundleInstructionId,
processBundleDescriptor.getPcollectionsMap(),
processBundleDescriptor.getCodersMap(),
processBundleDescriptor.getWindowingStrategiesMap(),
pCollectionIdsToConsumers,
Copy link
Contributor

@bsidhom bsidhom Aug 2, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How/when was pCollectionIdsToConsumers modified previously? I would like to know whether this is the right place to update processedPTransformIds.

Copy link
Contributor Author

@angoenka angoenka Aug 2, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

pCollectionIdsToConsumers is consumed at the time of bundle execution to lookup consumer for the pCollection.

beam/sdks/java/harness/src/main/java/org/apache/beam/fn/harness/DoFnPTransformRunnerFactory.java

Line 282 in 45ea59f

localNameToConsumerBuilder.putAll(

Copy link
Contributor

@bsidhom bsidhom Aug 3, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry, I meant when were elements added?

addStartFunction,
addFinishFunction,
splitListener);
processedPTransformIds.add(pTransformId);
}
}

public BeamFnApi.InstructionResponse.Builder processBundle(BeamFnApi.InstructionRequest request)
Expand All @@ -213,6 +215,7 @@ public BeamFnApi.InstructionResponse.Builder processBundle(BeamFnApi.Instruction
SetMultimap<String, String> pCollectionIdsToConsumingPTransforms = HashMultimap.create();
ListMultimap<String, FnDataReceiver<WindowedValue<?>>> pCollectionIdsToConsumers =
ArrayListMultimap.create();
HashSet<String> processedPTransformIds = new HashSet<>();
List<ThrowingRunnable> startFunctions = new ArrayList<>();
List<ThrowingRunnable> finishFunctions = new ArrayList<>();

Expand Down Expand Up @@ -271,6 +274,7 @@ public BeamFnApi.InstructionResponse.Builder processBundle(BeamFnApi.Instruction
bundleDescriptor,
pCollectionIdsToConsumingPTransforms,
pCollectionIdsToConsumers,
processedPTransformIds,
startFunctions::add,
finishFunctions::add,
splitListener);
Expand Down