You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ke...@apache.org on 2017/06/28 04:34:25 UTC
[1/6] beam git commit: Add utility to expand list of PCollectionViews
Repository: beam
Updated Branches:
refs/heads/master e5929bd13 -> 16f8000e2
Add utility to expand list of PCollectionViews
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/58fba590
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/58fba590
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/58fba590
Branch: refs/heads/master
Commit: 58fba590ddc554a343036a7beeffe9caa319aa81
Parents: e5929bd
Author: Kenneth Knowles <kl...@google.com>
Authored: Tue Jun 27 14:35:00 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Tue Jun 27 20:45:40 2017 -0700
----------------------------------------------------------------------
.../org/apache/beam/sdk/values/PCollectionViews.java | 14 ++++++++++++++
1 file changed, 14 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/58fba590/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionViews.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionViews.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionViews.java
index 0c04370..e17e146 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionViews.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/values/PCollectionViews.java
@@ -21,6 +21,7 @@ import com.google.common.base.Function;
import com.google.common.base.MoreObjects;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Multimap;
import java.io.IOException;
@@ -38,6 +39,7 @@ import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.IterableCoder;
import org.apache.beam.sdk.transforms.Materialization;
import org.apache.beam.sdk.transforms.Materializations;
+import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ViewFn;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.InvalidWindows;
@@ -139,6 +141,18 @@ public class PCollectionViews {
}
/**
+ * Expands a list of {@link PCollectionView} into the form needed for
+ * {@link PTransform#getAdditionalInputs()}.
+ */
+ public static Map<TupleTag<?>, PValue> toAdditionalInputs(Iterable<PCollectionView<?>> views) {
+ ImmutableMap.Builder<TupleTag<?>, PValue> additionalInputs = ImmutableMap.builder();
+ for (PCollectionView<?> view : views) {
+ additionalInputs.put(view.getTagInternal(), view.getPCollection());
+ }
+ return additionalInputs.build();
+ }
+
+ /**
* Implementation of conversion of singleton {@code Iterable<WindowedValue<T>>} to {@code T}.
*
* <p>For internal use only.
[6/6] beam git commit: This closes #3351: [BEAM-2371] Fix
getAdditionInputs for SplittableParDo transforms
Posted by ke...@apache.org.
This closes #3351: [BEAM-2371] Fix getAdditionInputs for SplittableParDo transforms
Use PCollectionViews.toAdditionalInputs in Combine
Use PCollectionViews.toAdditionalInputs in ParDo
Use PCollectionViews.toAdditionalInputs in ParDoMultiOverrideFactory
Fix getAdditionalInputs for SplittableParDo transforms
Add utility to expand list of PCollectionViews
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/16f8000e
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/16f8000e
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/16f8000e
Branch: refs/heads/master
Commit: 16f8000e24c5b56132987752b8e903392e5dd3e8
Parents: e5929bd 27674f0
Author: Kenneth Knowles <kl...@google.com>
Authored: Tue Jun 27 21:09:14 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Tue Jun 27 21:09:14 2017 -0700
----------------------------------------------------------------------
.../apache/beam/runners/apex/ApexRunner.java | 2 +-
.../core/construction/SplittableParDo.java | 66 +++++++++++++++-----
.../core/construction/SplittableParDoTest.java | 8 +--
.../direct/ParDoMultiOverrideFactory.java | 16 ++---
.../flink/FlinkStreamingPipelineTranslator.java | 2 +-
.../dataflow/SplittableParDoOverrides.java | 2 +-
.../org/apache/beam/sdk/transforms/Combine.java | 13 +---
.../org/apache/beam/sdk/transforms/ParDo.java | 14 +----
.../beam/sdk/values/PCollectionViews.java | 14 +++++
9 files changed, 79 insertions(+), 58 deletions(-)
----------------------------------------------------------------------
[2/6] beam git commit: Use PCollectionViews.toAdditionalInputs in
ParDoMultiOverrideFactory
Posted by ke...@apache.org.
Use PCollectionViews.toAdditionalInputs in ParDoMultiOverrideFactory
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/42382766
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/42382766
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/42382766
Branch: refs/heads/master
Commit: 423827665ae5923cd7fccc654bd9a5e1efed7876
Parents: a66bcd6
Author: Kenneth Knowles <kl...@google.com>
Authored: Tue Jun 27 14:39:06 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Tue Jun 27 21:08:10 2017 -0700
----------------------------------------------------------------------
.../runners/direct/ParDoMultiOverrideFactory.java | 14 +++-----------
1 file changed, 3 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/42382766/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoMultiOverrideFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoMultiOverrideFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoMultiOverrideFactory.java
index 9a26283..2904bc1 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoMultiOverrideFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoMultiOverrideFactory.java
@@ -19,7 +19,6 @@ package org.apache.beam.runners.direct;
import static com.google.common.base.Preconditions.checkState;
-import com.google.common.collect.ImmutableMap;
import java.util.List;
import java.util.Map;
import org.apache.beam.runners.core.KeyedWorkItem;
@@ -50,6 +49,7 @@ import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.PCollectionViews;
import org.apache.beam.sdk.values.PValue;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.TupleTagList;
@@ -123,11 +123,7 @@ class ParDoMultiOverrideFactory<InputT, OutputT>
@Override
public Map<TupleTag<?>, PValue> getAdditionalInputs() {
- ImmutableMap.Builder<TupleTag<?>, PValue> additionalInputs = ImmutableMap.builder();
- for (PCollectionView<?> sideInput : sideInputs) {
- additionalInputs.put(sideInput.getTagInternal(), sideInput.getPCollection());
- }
- return additionalInputs.build();
+ return PCollectionViews.toAdditionalInputs(sideInputs);
}
@Override
@@ -231,11 +227,7 @@ class ParDoMultiOverrideFactory<InputT, OutputT>
@Override
public Map<TupleTag<?>, PValue> getAdditionalInputs() {
- ImmutableMap.Builder<TupleTag<?>, PValue> additionalInputs = ImmutableMap.builder();
- for (PCollectionView<?> sideInput : sideInputs) {
- additionalInputs.put(sideInput.getTagInternal(), sideInput.getPCollection());
- }
- return additionalInputs.build();
+ return PCollectionViews.toAdditionalInputs(sideInputs);
}
@Override
[4/6] beam git commit: Use PCollectionViews.toAdditionalInputs in
ParDo
Posted by ke...@apache.org.
Use PCollectionViews.toAdditionalInputs in ParDo
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/ed476dd2
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/ed476dd2
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/ed476dd2
Branch: refs/heads/master
Commit: ed476dd2807577c8069087aa0764b21d1bb06512
Parents: 4238276
Author: Kenneth Knowles <kl...@google.com>
Authored: Tue Jun 27 14:41:30 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Tue Jun 27 21:08:11 2017 -0700
----------------------------------------------------------------------
.../java/org/apache/beam/sdk/transforms/ParDo.java | 14 +++-----------
1 file changed, 3 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/ed476dd2/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
index edf1419..db1f791 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ParDo.java
@@ -20,7 +20,6 @@ package org.apache.beam.sdk.transforms;
import static com.google.common.base.Preconditions.checkArgument;
import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
import java.io.Serializable;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
@@ -50,6 +49,7 @@ import org.apache.beam.sdk.util.SerializableUtils;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.PCollectionViews;
import org.apache.beam.sdk.values.PValue;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.TupleTagList;
@@ -662,11 +662,7 @@ public class ParDo {
*/
@Override
public Map<TupleTag<?>, PValue> getAdditionalInputs() {
- ImmutableMap.Builder<TupleTag<?>, PValue> additionalInputs = ImmutableMap.builder();
- for (PCollectionView<?> sideInput : sideInputs) {
- additionalInputs.put(sideInput.getTagInternal(), sideInput.getPCollection());
- }
- return additionalInputs.build();
+ return PCollectionViews.toAdditionalInputs(sideInputs);
}
}
@@ -807,11 +803,7 @@ public class ParDo {
*/
@Override
public Map<TupleTag<?>, PValue> getAdditionalInputs() {
- ImmutableMap.Builder<TupleTag<?>, PValue> additionalInputs = ImmutableMap.builder();
- for (PCollectionView<?> sideInput : sideInputs) {
- additionalInputs.put(sideInput.getTagInternal(), sideInput.getPCollection());
- }
- return additionalInputs.build();
+ return PCollectionViews.toAdditionalInputs(sideInputs);
}
}
[5/6] beam git commit: Use PCollectionViews.toAdditionalInputs in
Combine
Posted by ke...@apache.org.
Use PCollectionViews.toAdditionalInputs in Combine
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/27674f07
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/27674f07
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/27674f07
Branch: refs/heads/master
Commit: 27674f07cf8363bb6b3c051a990caa5d61b8cd5c
Parents: ed476dd
Author: Kenneth Knowles <kl...@google.com>
Authored: Tue Jun 27 14:44:50 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Tue Jun 27 21:08:11 2017 -0700
----------------------------------------------------------------------
.../java/org/apache/beam/sdk/transforms/Combine.java | 13 ++-----------
1 file changed, 2 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/27674f07/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java
index 6a90bcf..d7effb5 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Combine.java
@@ -20,7 +20,6 @@ package org.apache.beam.sdk.transforms;
import static com.google.common.base.Preconditions.checkState;
import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import java.io.IOException;
import java.io.InputStream;
@@ -1122,11 +1121,7 @@ public class Combine {
*/
@Override
public Map<TupleTag<?>, PValue> getAdditionalInputs() {
- ImmutableMap.Builder<TupleTag<?>, PValue> additionalInputs = ImmutableMap.builder();
- for (PCollectionView<?> sideInput : sideInputs) {
- additionalInputs.put(sideInput.getTagInternal(), sideInput.getPCollection());
- }
- return additionalInputs.build();
+ return PCollectionViews.toAdditionalInputs(sideInputs);
}
/**
@@ -1578,11 +1573,7 @@ public class Combine {
*/
@Override
public Map<TupleTag<?>, PValue> getAdditionalInputs() {
- ImmutableMap.Builder<TupleTag<?>, PValue> additionalInputs = ImmutableMap.builder();
- for (PCollectionView<?> sideInput : sideInputs) {
- additionalInputs.put(sideInput.getTagInternal(), sideInput.getPCollection());
- }
- return additionalInputs.build();
+ return PCollectionViews.toAdditionalInputs(sideInputs);
}
@Override
[3/6] beam git commit: Fix getAdditionalInputs for SplittableParDo
transforms
Posted by ke...@apache.org.
Fix getAdditionalInputs for SplittableParDo transforms
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/a66bcd68
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/a66bcd68
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/a66bcd68
Branch: refs/heads/master
Commit: a66bcd68a1e56d5d38fccfce2ffeec28ba1c82de
Parents: 58fba59
Author: Kenneth Knowles <kl...@google.com>
Authored: Tue Jun 13 10:00:09 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Tue Jun 27 21:08:10 2017 -0700
----------------------------------------------------------------------
.../apache/beam/runners/apex/ApexRunner.java | 2 +-
.../core/construction/SplittableParDo.java | 66 +++++++++++++++-----
.../core/construction/SplittableParDoTest.java | 8 +--
.../direct/ParDoMultiOverrideFactory.java | 2 +-
.../flink/FlinkStreamingPipelineTranslator.java | 2 +-
.../dataflow/SplittableParDoOverrides.java | 2 +-
6 files changed, 57 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/a66bcd68/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunner.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunner.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunner.java
index 95b354a..fd0a1c9 100644
--- a/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunner.java
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunner.java
@@ -381,7 +381,7 @@ public class ApexRunner extends PipelineRunner<ApexRunnerResult> {
AppliedPTransform<PCollection<InputT>, PCollectionTuple, MultiOutput<InputT, OutputT>>
transform) {
return PTransformReplacement.of(PTransformReplacements.getSingletonMainInput(transform),
- new SplittableParDo<>(transform.getTransform()));
+ SplittableParDo.forJavaParDo(transform.getTransform()));
}
@Override
http://git-wip-us.apache.org/repos/asf/beam/blob/a66bcd68/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/SplittableParDo.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/SplittableParDo.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/SplittableParDo.java
index 5ccafcb..f31b495 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/SplittableParDo.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/SplittableParDo.java
@@ -18,9 +18,9 @@
package org.apache.beam.runners.core.construction;
import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
import java.util.List;
+import java.util.Map;
import java.util.UUID;
import org.apache.beam.runners.core.construction.PTransformTranslation.RawPTransform;
import org.apache.beam.sdk.annotations.Experimental;
@@ -40,6 +40,8 @@ import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.PCollectionViews;
+import org.apache.beam.sdk.values.PValue;
import org.apache.beam.sdk.values.TupleTag;
import org.apache.beam.sdk.values.TupleTagList;
import org.apache.beam.sdk.values.WindowingStrategy;
@@ -64,7 +66,11 @@ import org.apache.beam.sdk.values.WindowingStrategy;
@Experimental(Experimental.Kind.SPLITTABLE_DO_FN)
public class SplittableParDo<InputT, OutputT, RestrictionT>
extends PTransform<PCollection<InputT>, PCollectionTuple> {
- private final ParDo.MultiOutput<InputT, OutputT> parDo;
+
+ private final DoFn<InputT, OutputT> doFn;
+ private final List<PCollectionView<?>> sideInputs;
+ private final TupleTag<OutputT> mainOutputTag;
+ private final TupleTagList additionalOutputTags;
public static final String SPLITTABLE_PROCESS_URN =
"urn:beam:runners_core:transforms:splittable_process:v1";
@@ -75,24 +81,39 @@ public class SplittableParDo<InputT, OutputT, RestrictionT>
public static final String SPLITTABLE_GBKIKWI_URN =
"urn:beam:runners_core:transforms:splittable_gbkikwi:v1";
+ private SplittableParDo(
+ DoFn<InputT, OutputT> doFn,
+ TupleTag<OutputT> mainOutputTag,
+ List<PCollectionView<?>> sideInputs,
+ TupleTagList additionalOutputTags) {
+ checkArgument(
+ DoFnSignatures.getSignature(doFn.getClass()).processElement().isSplittable(),
+ "fn must be a splittable DoFn");
+ this.doFn = doFn;
+ this.mainOutputTag = mainOutputTag;
+ this.sideInputs = sideInputs;
+ this.additionalOutputTags = additionalOutputTags;
+ }
+
/**
- * Creates the transform for the given original multi-output {@link ParDo}.
+ * Creates a {@link SplittableParDo} from an original Java {@link ParDo}.
*
* @param parDo The splittable {@link ParDo} transform.
*/
- public SplittableParDo(ParDo.MultiOutput<InputT, OutputT> parDo) {
- checkNotNull(parDo, "parDo must not be null");
- this.parDo = parDo;
- checkArgument(
- DoFnSignatures.getSignature(parDo.getFn().getClass()).processElement().isSplittable(),
- "fn must be a splittable DoFn");
+ public static <InputT, OutputT> SplittableParDo<InputT, OutputT, ?> forJavaParDo(
+ ParDo.MultiOutput<InputT, OutputT> parDo) {
+ checkArgument(parDo != null, "parDo must not be null");
+ return new SplittableParDo(
+ parDo.getFn(),
+ parDo.getMainOutputTag(),
+ parDo.getSideInputs(),
+ parDo.getAdditionalOutputTags());
}
@Override
public PCollectionTuple expand(PCollection<InputT> input) {
- DoFn<InputT, OutputT> fn = parDo.getFn();
Coder<RestrictionT> restrictionCoder =
- DoFnInvokers.invokerFor(fn)
+ DoFnInvokers.invokerFor(doFn)
.invokeGetRestrictionCoder(input.getPipeline().getCoderRegistry());
Coder<KV<InputT, RestrictionT>> splitCoder = KvCoder.of(input.getCoder(), restrictionCoder);
@@ -100,9 +121,10 @@ public class SplittableParDo<InputT, OutputT, RestrictionT>
input
.apply(
"Pair with initial restriction",
- ParDo.of(new PairWithRestrictionFn<InputT, OutputT, RestrictionT>(fn)))
+ ParDo.of(new PairWithRestrictionFn<InputT, OutputT, RestrictionT>(doFn)))
.setCoder(splitCoder)
- .apply("Split restriction", ParDo.of(new SplitRestrictionFn<InputT, RestrictionT>(fn)))
+ .apply(
+ "Split restriction", ParDo.of(new SplitRestrictionFn<InputT, RestrictionT>(doFn)))
.setCoder(splitCoder)
// ProcessFn requires all input elements to be in a single window and have a single
// element per work item. This must precede the unique keying so each key has a single
@@ -115,13 +137,18 @@ public class SplittableParDo<InputT, OutputT, RestrictionT>
return keyedRestrictions.apply(
"ProcessKeyedElements",
new ProcessKeyedElements<>(
- fn,
+ doFn,
input.getCoder(),
restrictionCoder,
(WindowingStrategy<InputT, ?>) input.getWindowingStrategy(),
- parDo.getSideInputs(),
- parDo.getMainOutputTag(),
- parDo.getAdditionalOutputTags()));
+ sideInputs,
+ mainOutputTag,
+ additionalOutputTags));
+ }
+
+ @Override
+ public Map<TupleTag<?>, PValue> getAdditionalInputs() {
+ return PCollectionViews.toAdditionalInputs(sideInputs);
}
/**
@@ -231,6 +258,11 @@ public class SplittableParDo<InputT, OutputT, RestrictionT>
}
@Override
+ public Map<TupleTag<?>, PValue> getAdditionalInputs() {
+ return PCollectionViews.toAdditionalInputs(sideInputs);
+ }
+
+ @Override
public String getUrn() {
return SPLITTABLE_PROCESS_KEYED_ELEMENTS_URN;
}
http://git-wip-us.apache.org/repos/asf/beam/blob/a66bcd68/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/SplittableParDoTest.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/SplittableParDoTest.java b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/SplittableParDoTest.java
index 6e4d6c4..f4c596e 100644
--- a/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/SplittableParDoTest.java
+++ b/runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/SplittableParDoTest.java
@@ -122,14 +122,14 @@ public class SplittableParDoTest {
"Applying a bounded SDF to a bounded collection produces a bounded collection",
PCollection.IsBounded.BOUNDED,
makeBoundedCollection(pipeline)
- .apply("bounded to bounded", new SplittableParDo<>(makeParDo(boundedFn)))
+ .apply("bounded to bounded", SplittableParDo.forJavaParDo(makeParDo(boundedFn)))
.get(MAIN_OUTPUT_TAG)
.isBounded());
assertEquals(
"Applying a bounded SDF to an unbounded collection produces an unbounded collection",
PCollection.IsBounded.UNBOUNDED,
makeUnboundedCollection(pipeline)
- .apply("bounded to unbounded", new SplittableParDo<>(makeParDo(boundedFn)))
+ .apply("bounded to unbounded", SplittableParDo.forJavaParDo(makeParDo(boundedFn)))
.get(MAIN_OUTPUT_TAG)
.isBounded());
}
@@ -143,14 +143,14 @@ public class SplittableParDoTest {
"Applying an unbounded SDF to a bounded collection produces a bounded collection",
PCollection.IsBounded.UNBOUNDED,
makeBoundedCollection(pipeline)
- .apply("unbounded to bounded", new SplittableParDo<>(makeParDo(unboundedFn)))
+ .apply("unbounded to bounded", SplittableParDo.forJavaParDo(makeParDo(unboundedFn)))
.get(MAIN_OUTPUT_TAG)
.isBounded());
assertEquals(
"Applying an unbounded SDF to an unbounded collection produces an unbounded collection",
PCollection.IsBounded.UNBOUNDED,
makeUnboundedCollection(pipeline)
- .apply("unbounded to unbounded", new SplittableParDo<>(makeParDo(unboundedFn)))
+ .apply("unbounded to unbounded", SplittableParDo.forJavaParDo(makeParDo(unboundedFn)))
.get(MAIN_OUTPUT_TAG)
.isBounded());
}
http://git-wip-us.apache.org/repos/asf/beam/blob/a66bcd68/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoMultiOverrideFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoMultiOverrideFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoMultiOverrideFactory.java
index b20113e..9a26283 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoMultiOverrideFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoMultiOverrideFactory.java
@@ -81,7 +81,7 @@ class ParDoMultiOverrideFactory<InputT, OutputT>
DoFn<InputT, OutputT> fn = transform.getFn();
DoFnSignature signature = DoFnSignatures.getSignature(fn.getClass());
if (signature.processElement().isSplittable()) {
- return new SplittableParDo(transform);
+ return (PTransform) SplittableParDo.forJavaParDo(transform);
} else if (signature.stateDeclarations().size() > 0
|| signature.timerDeclarations().size() > 0) {
http://git-wip-us.apache.org/repos/asf/beam/blob/a66bcd68/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPipelineTranslator.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPipelineTranslator.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPipelineTranslator.java
index 27bb4ec..ebc9345 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPipelineTranslator.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkStreamingPipelineTranslator.java
@@ -188,7 +188,7 @@ class FlinkStreamingPipelineTranslator extends FlinkPipelineTranslator {
transform) {
return PTransformReplacement.of(
PTransformReplacements.getSingletonMainInput(transform),
- new SplittableParDo<>(transform.getTransform()));
+ SplittableParDo.forJavaParDo(transform.getTransform()));
}
@Override
http://git-wip-us.apache.org/repos/asf/beam/blob/a66bcd68/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/SplittableParDoOverrides.java
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/SplittableParDoOverrides.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/SplittableParDoOverrides.java
index 9322878..fc010f8 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/SplittableParDoOverrides.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/SplittableParDoOverrides.java
@@ -64,7 +64,7 @@ class SplittableParDoOverrides {
appliedTransform) {
return PTransformReplacement.of(
PTransformReplacements.getSingletonMainInput(appliedTransform),
- new SplittableParDo<>(appliedTransform.getTransform()));
+ SplittableParDo.forJavaParDo(appliedTransform.getTransform()));
}
@Override