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/05/26 04:08:07 UTC
[01/12] beam git commit: Add URN for Splittable ProcessElement
pseudo-primitive
Repository: beam
Updated Branches:
refs/heads/master 50200f23a -> 9cd12907e
Add URN for Splittable ProcessElement pseudo-primitive
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/877408e0
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/877408e0
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/877408e0
Branch: refs/heads/master
Commit: 877408e078b783b3a782ae1ba0e335ea44c4a0e2
Parents: 1dc134c
Author: Kenneth Knowles <kl...@google.com>
Authored: Wed May 24 15:28:38 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu May 25 11:14:29 2017 -0700
----------------------------------------------------------------------
.../apache/beam/runners/core/construction/SplittableParDo.java | 3 +++
1 file changed, 3 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/877408e0/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 23ba66a..dfca7d2 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
@@ -64,6 +64,9 @@ public class SplittableParDo<InputT, OutputT, RestrictionT>
extends PTransform<PCollection<InputT>, PCollectionTuple> {
private final ParDo.MultiOutput<InputT, OutputT> parDo;
+ public static final String SPLITTABLE_PROCESS_URN =
+ "urn:beam:runners_core:transforms:splittable_process:v1";
+
/**
* Creates the transform for the given original multi-output {@link ParDo}.
*
[04/12] beam git commit: Allow getting URN for class of transform via
translator
Posted by ke...@apache.org.
Allow getting URN for class of transform via translator
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/0bf4ddbe
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/0bf4ddbe
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/0bf4ddbe
Branch: refs/heads/master
Commit: 0bf4ddbea7d0f790fad0cc8df20f0d01f38dd568
Parents: 877408e
Author: Kenneth Knowles <kl...@google.com>
Authored: Fri May 19 20:55:25 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu May 25 11:16:43 2017 -0700
----------------------------------------------------------------------
.../core/construction/PTransformTranslation.java | 14 +++++++++++++-
.../runners/core/construction/ParDoTranslation.java | 5 +++++
.../core/construction/WindowIntoTranslation.java | 7 +++++++
3 files changed, 25 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/0bf4ddbe/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java
index 8be023a..35bb0e3 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java
@@ -116,11 +116,23 @@ public class PTransformTranslation {
return tag.getId();
}
+ public static String urnForTransform(PTransform<?, ?> transform) {
+ TransformPayloadTranslator translator =
+ KNOWN_PAYLOAD_TRANSLATORS.get(transform.getClass());
+ if (translator == null) {
+ throw new IllegalStateException(
+ String.format("No translator known for %s", transform.getClass().getName()));
+ }
+
+ return translator.getUrn(transform);
+ }
+
/**
* A translator consumes a {@link PTransform} application and produces the appropriate
* FunctionSpec for a distinguished or primitive transform within the Beam runner API.
*/
public interface TransformPayloadTranslator<T extends PTransform<?, ?>> {
- FunctionSpec translate(AppliedPTransform<?, ?, T> transform, SdkComponents components);
+ String getUrn(T transform);
+ FunctionSpec translate(AppliedPTransform<?, ?, T> application, SdkComponents components);
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/0bf4ddbe/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ParDoTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ParDoTranslation.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ParDoTranslation.java
index 83277bb..1c81f8c 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ParDoTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ParDoTranslation.java
@@ -101,6 +101,11 @@ public class ParDoTranslation {
private ParDoPayloadTranslator() {}
@Override
+ public String getUrn(ParDo.MultiOutput<?, ?> transform) {
+ return PAR_DO_TRANSFORM_URN;
+ }
+
+ @Override
public FunctionSpec translate(
AppliedPTransform<?, ?, MultiOutput<?, ?>> transform, SdkComponents components) {
ParDoPayload payload = toProto(transform.getTransform(), components);
http://git-wip-us.apache.org/repos/asf/beam/blob/0bf4ddbe/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowIntoTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowIntoTranslation.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowIntoTranslation.java
index 215beba..33faa02 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowIntoTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowIntoTranslation.java
@@ -27,6 +27,7 @@ import org.apache.beam.sdk.common.runner.v1.RunnerApi.SdkFunctionSpec;
import org.apache.beam.sdk.common.runner.v1.RunnerApi.WindowIntoPayload;
import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.transforms.windowing.Window.Assign;
import org.apache.beam.sdk.transforms.windowing.WindowFn;
/**
@@ -36,6 +37,12 @@ import org.apache.beam.sdk.transforms.windowing.WindowFn;
public class WindowIntoTranslation {
static class WindowAssignTranslator implements TransformPayloadTranslator<Window.Assign<?>> {
+
+ @Override
+ public String getUrn(Assign<?> transform) {
+ return PTransforms.WINDOW_TRANSFORM_URN;
+ }
+
@Override
public FunctionSpec translate(
AppliedPTransform<?, ?, Window.Assign<?>> transform, SdkComponents components) {
[12/12] beam git commit: This closes #3195: [BEAM-2333] Key
DirectRunner translators off URN
Posted by ke...@apache.org.
This closes #3195: [BEAM-2333] Key DirectRunner translators off URN
URNs for DirectRunner TransformEvaluator and RootInputProvider
Add RawPTransform, which can just vend its URN and payload
Add transform-analysis helpers to ReadTranslation
Add trivial FlattenTranslator to access URN
Register WindowIntoTranslator
Register ReadTranslator
Add registration for Read and WindowInto translators
Allow getting URN for class of transform via translator
Add URN for Splittable ProcessElement pseudo-primitive
Centralize primitive URNs in PTransformTranslation class
Make SdkComponents public for TransformPayloadTranslator
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/9cd12907
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/9cd12907
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/9cd12907
Branch: refs/heads/master
Commit: 9cd12907ecbcf19b372185dff70a21e7e11fff6c
Parents: 50200f2 0e29cc5
Author: Kenneth Knowles <kl...@google.com>
Authored: Thu May 25 20:52:13 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu May 25 20:52:13 2017 -0700
----------------------------------------------------------------------
.../core/construction/FlattenTranslator.java | 63 +++++++++
.../construction/PCollectionTranslation.java | 4 +-
.../construction/PTransformTranslation.java | 72 ++++++++++-
.../core/construction/ParDoTranslation.java | 14 +-
.../core/construction/ReadTranslation.java | 117 ++++++++++++++++-
.../core/construction/SdkComponents.java | 2 +-
.../core/construction/SplittableParDo.java | 3 +
.../construction/WindowIntoTranslation.java | 48 +++++++
runners/direct-java/pom.xml | 5 +
.../direct/BoundedReadEvaluatorFactory.java | 14 +-
.../beam/runners/direct/DirectGroupByKey.java | 21 ++-
.../beam/runners/direct/EmptyInputProvider.java | 8 +-
.../direct/ParDoMultiOverrideFactory.java | 13 +-
.../runners/direct/ReadEvaluatorFactory.java | 97 ++++++++++++++
.../beam/runners/direct/RootInputProvider.java | 7 +-
.../runners/direct/RootProviderRegistry.java | 28 ++--
.../apache/beam/runners/direct/SourceShard.java | 33 +++++
.../direct/TestStreamEvaluatorFactory.java | 28 ++--
.../direct/TransformEvaluatorRegistry.java | 128 ++++++++++++++-----
.../direct/UnboundedReadEvaluatorFactory.java | 31 +++--
.../runners/direct/ViewOverrideFactory.java | 12 +-
.../src/main/resources/beam/findbugs-filter.xml | 7 +
22 files changed, 656 insertions(+), 99 deletions(-)
----------------------------------------------------------------------
[03/12] beam git commit: Centralize primitive URNs in
PTransformTranslation class
Posted by ke...@apache.org.
Centralize primitive URNs in PTransformTranslation class
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/1dc134cf
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/1dc134cf
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/1dc134cf
Branch: refs/heads/master
Commit: 1dc134cf341d1eb4fa936f6fe7b83a3edbb64687
Parents: 26c9368
Author: Kenneth Knowles <kl...@google.com>
Authored: Fri May 19 20:54:36 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu May 25 11:14:29 2017 -0700
----------------------------------------------------------------------
.../runners/core/construction/PTransformTranslation.java | 7 +++++++
.../beam/runners/core/construction/ParDoTranslation.java | 9 +++------
2 files changed, 10 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/1dc134cf/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java
index 86638de..8be023a 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java
@@ -38,6 +38,13 @@ import org.apache.beam.sdk.values.TupleTag;
* protocol buffers}.
*/
public class PTransformTranslation {
+
+ public static final String PAR_DO_TRANSFORM_URN = "urn:beam:transform:pardo:v1";
+ public static final String FLATTEN_TRANSFORM_URN = "urn:beam:transform:flatten:v1";
+ public static final String GROUP_BY_KEY_TRANSFORM_URN = "urn:beam:transform:groupbykey:v1";
+ public static final String READ_TRANSFORM_URN = "urn:beam:transform:read:v1";
+ public static final String WINDOW_TRANSFORM_URN = "urn:beam:transform:window:v1";
+
private static final Map<Class<? extends PTransform>, TransformPayloadTranslator>
KNOWN_PAYLOAD_TRANSLATORS = loadTransformPayloadTranslators();
http://git-wip-us.apache.org/repos/asf/beam/blob/1dc134cf/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ParDoTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ParDoTranslation.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ParDoTranslation.java
index 28d577f..83277bb 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ParDoTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ParDoTranslation.java
@@ -19,6 +19,7 @@
package org.apache.beam.runners.core.construction;
import static com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.runners.core.construction.PTransformTranslation.PAR_DO_TRANSFORM_URN;
import com.google.auto.service.AutoService;
import com.google.auto.value.AutoValue;
@@ -75,10 +76,6 @@ import org.apache.beam.sdk.values.WindowingStrategy;
*/
public class ParDoTranslation {
/**
- * The URN for a {@link ParDoPayload}.
- */
- public static final String PAR_DO_PAYLOAD_URN = "urn:beam:pardo:v1";
- /**
* The URN for an unknown Java {@link DoFn}.
*/
public static final String CUSTOM_JAVA_DO_FN_URN = "urn:beam:dofn:javasdk:0.1";
@@ -108,7 +105,7 @@ public class ParDoTranslation {
AppliedPTransform<?, ?, MultiOutput<?, ?>> transform, SdkComponents components) {
ParDoPayload payload = toProto(transform.getTransform(), components);
return RunnerApi.FunctionSpec.newBuilder()
- .setUrn(PAR_DO_PAYLOAD_URN)
+ .setUrn(PAR_DO_TRANSFORM_URN)
.setParameter(Any.pack(payload))
.build();
}
@@ -166,7 +163,7 @@ public class ParDoTranslation {
public static RunnerApi.PCollection getMainInput(
RunnerApi.PTransform ptransform, Components components) throws IOException {
checkArgument(
- ptransform.getSpec().getUrn().equals(PAR_DO_PAYLOAD_URN),
+ ptransform.getSpec().getUrn().equals(PAR_DO_TRANSFORM_URN),
"Unexpected payload type %s",
ptransform.getSpec().getUrn());
ParDoPayload payload = ptransform.getSpec().getParameter().unpack(ParDoPayload.class);
[05/12] beam git commit: Register ReadTranslator
Posted by ke...@apache.org.
Register ReadTranslator
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/d91c840f
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/d91c840f
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/d91c840f
Branch: refs/heads/master
Commit: d91c840f009bdfda14f8668172d076a7e9f12f5e
Parents: afaebb1
Author: Kenneth Knowles <kl...@google.com>
Authored: Mon May 22 21:09:20 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu May 25 11:16:47 2017 -0700
----------------------------------------------------------------------
.../beam/runners/core/construction/ReadTranslation.java | 8 ++++----
1 file changed, 4 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/d91c840f/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
index aff5fc9..3ddde8d 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
@@ -143,7 +143,7 @@ public class ReadTranslation {
@Override
public String getUrn(Read.Unbounded<?> transform) {
- return PTransformTranslation.WINDOW_TRANSFORM_URN;
+ return PTransformTranslation.READ_TRANSFORM_URN;
}
@Override
@@ -151,7 +151,7 @@ public class ReadTranslation {
AppliedPTransform<?, ?, Read.Unbounded<?>> transform, SdkComponents components) {
ReadPayload payload = toProto(transform.getTransform());
return RunnerApi.FunctionSpec.newBuilder()
- .setUrn(PTransformTranslation.READ_TRANSFORM_URN)
+ .setUrn(getUrn(transform.getTransform()))
.setParameter(Any.pack(payload))
.build();
}
@@ -170,7 +170,7 @@ public class ReadTranslation {
@Override
public String getUrn(Read.Bounded<?> transform) {
- return PTransformTranslation.WINDOW_TRANSFORM_URN;
+ return PTransformTranslation.READ_TRANSFORM_URN;
}
@Override
@@ -178,7 +178,7 @@ public class ReadTranslation {
AppliedPTransform<?, ?, Read.Bounded<?>> transform, SdkComponents components) {
ReadPayload payload = toProto(transform.getTransform());
return RunnerApi.FunctionSpec.newBuilder()
- .setUrn(PTransformTranslation.READ_TRANSFORM_URN)
+ .setUrn(getUrn(transform.getTransform()))
.setParameter(Any.pack(payload))
.build();
}
[09/12] beam git commit: Add transform-analysis helpers to
ReadTranslation
Posted by ke...@apache.org.
Add transform-analysis helpers to ReadTranslation
These helpers allow a runner to extract a source and inspect boundedness
without reference to the specific user-facing Java classes Read.Bounded
and Read.Unbounded.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/8e09596a
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/8e09596a
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/8e09596a
Branch: refs/heads/master
Commit: 8e09596a981fe69edb7a1560e864bc852978ba81
Parents: 121631c
Author: Kenneth Knowles <kl...@google.com>
Authored: Tue May 23 21:15:48 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu May 25 11:17:04 2017 -0700
----------------------------------------------------------------------
.../construction/PCollectionTranslation.java | 4 +-
.../core/construction/ReadTranslation.java | 43 ++++++++++++++++++++
2 files changed, 45 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/8e09596a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PCollectionTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PCollectionTranslation.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PCollectionTranslation.java
index 303c02d..968966f 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PCollectionTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PCollectionTranslation.java
@@ -64,7 +64,7 @@ public class PCollectionTranslation {
components.getWindowingStrategiesOrThrow(pCollection.getWindowingStrategyId()), components);
}
- private static RunnerApi.IsBounded toProto(IsBounded bounded) {
+ static RunnerApi.IsBounded toProto(IsBounded bounded) {
switch (bounded) {
case BOUNDED:
return RunnerApi.IsBounded.BOUNDED;
@@ -76,7 +76,7 @@ public class PCollectionTranslation {
}
}
- private static IsBounded fromProto(RunnerApi.IsBounded isBounded) {
+ static IsBounded fromProto(RunnerApi.IsBounded isBounded) {
switch (isBounded) {
case BOUNDED:
return IsBounded.BOUNDED;
http://git-wip-us.apache.org/repos/asf/beam/blob/8e09596a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
index 3ddde8d..572384b 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
@@ -26,6 +26,8 @@ import com.google.protobuf.Any;
import com.google.protobuf.ByteString;
import com.google.protobuf.BytesValue;
import com.google.protobuf.InvalidProtocolBufferException;
+import java.io.IOException;
+import java.util.Collections;
import java.util.Map;
import org.apache.beam.runners.core.construction.PTransformTranslation.TransformPayloadTranslator;
import org.apache.beam.sdk.common.runner.v1.RunnerApi;
@@ -40,6 +42,8 @@ import org.apache.beam.sdk.io.UnboundedSource;
import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.util.SerializableUtils;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
/**
* Methods for translating {@link Read.Bounded} and {@link Read.Unbounded}
@@ -102,6 +106,29 @@ public class ReadTranslation {
"BoundedSource");
}
+ public static <T> BoundedSource<T> boundedSourceFromTransform(
+ AppliedPTransform<PBegin, PCollection<T>, PTransform<PBegin, PCollection<T>>> transform)
+ throws IOException {
+ return (BoundedSource<T>) boundedSourceFromProto(getReadPayload(transform));
+ }
+
+ public static <T, CheckpointT extends UnboundedSource.CheckpointMark>
+ UnboundedSource<T, CheckpointT> unboundedSourceFromTransform(
+ AppliedPTransform<PBegin, PCollection<T>, PTransform<PBegin, PCollection<T>>> transform)
+ throws IOException {
+ return (UnboundedSource<T, CheckpointT>) unboundedSourceFromProto(getReadPayload(transform));
+ }
+
+ private static <T> ReadPayload getReadPayload(
+ AppliedPTransform<PBegin, PCollection<T>, PTransform<PBegin, PCollection<T>>> transform)
+ throws IOException {
+ return PTransformTranslation.toProto(
+ transform, Collections.<AppliedPTransform<?, ?, ?>>emptyList(), SdkComponents.create())
+ .getSpec()
+ .getParameter()
+ .unpack(ReadPayload.class);
+ }
+
private static SdkFunctionSpec toProto(UnboundedSource<?, ?> source) {
return SdkFunctionSpec.newBuilder()
.setSpec(
@@ -130,6 +157,22 @@ public class ReadTranslation {
"BoundedSource");
}
+ public static PCollection.IsBounded sourceIsBounded(AppliedPTransform<?, ?, ?> transform) {
+ try {
+ return PCollectionTranslation.fromProto(
+ PTransformTranslation.toProto(
+ transform,
+ Collections.<AppliedPTransform<?, ?, ?>>emptyList(),
+ SdkComponents.create())
+ .getSpec()
+ .getParameter()
+ .unpack(ReadPayload.class)
+ .getIsBounded());
+ } catch (IOException e) {
+ throw new RuntimeException("Internal error determining boundedness of Read", e);
+ }
+ }
+
/**
* A {@link TransformPayloadTranslator} for {@link Read.Unbounded}.
*/
[10/12] beam git commit: URNs for DirectRunner TransformEvaluator and
RootInputProvider
Posted by ke...@apache.org.
URNs for DirectRunner TransformEvaluator and RootInputProvider
This makes two of the Java DirectRunner's registries key off URN instead
of Java class. A root TransformEvaluator requires shards generated by
its associated RootInputProvider, hence changing both at once.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/0e29cc52
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/0e29cc52
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/0e29cc52
Branch: refs/heads/master
Commit: 0e29cc52a3e4b0d9ae5ff3907f10e4e87b734186
Parents: 663ad88
Author: Kenneth Knowles <kl...@google.com>
Authored: Fri May 19 20:53:32 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu May 25 11:17:08 2017 -0700
----------------------------------------------------------------------
runners/direct-java/pom.xml | 5 +
.../direct/BoundedReadEvaluatorFactory.java | 14 +-
.../beam/runners/direct/DirectGroupByKey.java | 21 ++-
.../beam/runners/direct/EmptyInputProvider.java | 8 +-
.../direct/ParDoMultiOverrideFactory.java | 13 +-
.../runners/direct/ReadEvaluatorFactory.java | 97 ++++++++++++++
.../beam/runners/direct/RootInputProvider.java | 7 +-
.../runners/direct/RootProviderRegistry.java | 28 ++--
.../apache/beam/runners/direct/SourceShard.java | 33 +++++
.../direct/TestStreamEvaluatorFactory.java | 28 ++--
.../direct/TransformEvaluatorRegistry.java | 128 ++++++++++++++-----
.../direct/UnboundedReadEvaluatorFactory.java | 31 +++--
.../runners/direct/ViewOverrideFactory.java | 12 +-
.../src/main/resources/beam/findbugs-filter.xml | 7 +
14 files changed, 344 insertions(+), 88 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/0e29cc52/runners/direct-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/direct-java/pom.xml b/runners/direct-java/pom.xml
index bec2113..cba4b09 100644
--- a/runners/direct-java/pom.xml
+++ b/runners/direct-java/pom.xml
@@ -208,6 +208,11 @@
</dependency>
<dependency>
+ <groupId>com.google.protobuf</groupId>
+ <artifactId>protobuf-java</artifactId>
+ </dependency>
+
+ <dependency>
<groupId>joda-time</groupId>
<artifactId>joda-time</artifactId>
</dependency>
http://git-wip-us.apache.org/repos/asf/beam/blob/0e29cc52/runners/direct-java/src/main/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactory.java
index 76db861..fcaaa84 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/BoundedReadEvaluatorFactory.java
@@ -33,10 +33,10 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import javax.annotation.Nullable;
+import org.apache.beam.runners.core.construction.ReadTranslation;
import org.apache.beam.runners.direct.StepTransformResult.Builder;
import org.apache.beam.sdk.io.BoundedSource;
import org.apache.beam.sdk.io.BoundedSource.BoundedReader;
-import org.apache.beam.sdk.io.Read;
import org.apache.beam.sdk.io.Read.Bounded;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.runners.AppliedPTransform;
@@ -180,16 +180,17 @@ final class BoundedReadEvaluatorFactory implements TransformEvaluatorFactory {
}
@AutoValue
- abstract static class BoundedSourceShard<T> {
+ abstract static class BoundedSourceShard<T> implements SourceShard<T> {
static <T> BoundedSourceShard<T> of(BoundedSource<T> source) {
return new AutoValue_BoundedReadEvaluatorFactory_BoundedSourceShard<>(source);
}
- abstract BoundedSource<T> getSource();
+ @Override
+ public abstract BoundedSource<T> getSource();
}
static class InputProvider<T>
- implements RootInputProvider<T, BoundedSourceShard<T>, PBegin, Read.Bounded<T>> {
+ implements RootInputProvider<T, BoundedSourceShard<T>, PBegin> {
private final EvaluationContext evaluationContext;
InputProvider(EvaluationContext evaluationContext) {
@@ -198,9 +199,10 @@ final class BoundedReadEvaluatorFactory implements TransformEvaluatorFactory {
@Override
public Collection<CommittedBundle<BoundedSourceShard<T>>> getInitialInputs(
- AppliedPTransform<PBegin, PCollection<T>, Read.Bounded<T>> transform, int targetParallelism)
+ AppliedPTransform<PBegin, PCollection<T>, PTransform<PBegin, PCollection<T>>> transform,
+ int targetParallelism)
throws Exception {
- BoundedSource<T> source = transform.getTransform().getSource();
+ BoundedSource<T> source = ReadTranslation.boundedSourceFromTransform(transform);
PipelineOptions options = evaluationContext.getPipelineOptions();
long estimatedBytes = source.getEstimatedSizeBytes(options);
long bytesPerBundle = estimatedBytes / targetParallelism;
http://git-wip-us.apache.org/repos/asf/beam/blob/0e29cc52/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectGroupByKey.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectGroupByKey.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectGroupByKey.java
index 791615a..f239070 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectGroupByKey.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/DirectGroupByKey.java
@@ -20,9 +20,11 @@ package org.apache.beam.runners.direct;
import static com.google.common.base.Preconditions.checkArgument;
+import com.google.protobuf.Message;
import org.apache.beam.runners.core.KeyedWorkItem;
import org.apache.beam.runners.core.KeyedWorkItemCoder;
import org.apache.beam.runners.core.construction.ForwardingPTransform;
+import org.apache.beam.runners.core.construction.PTransformTranslation;
import org.apache.beam.sdk.coders.CannotProvideCoderException;
import org.apache.beam.sdk.coders.Coder;
import org.apache.beam.sdk.coders.IterableCoder;
@@ -37,6 +39,9 @@ class DirectGroupByKey<K, V>
extends ForwardingPTransform<PCollection<KV<K, V>>, PCollection<KV<K, Iterable<V>>>> {
private final GroupByKey<K, V> original;
+ static final String DIRECT_GBKO_URN = "urn:beam:directrunner:transforms:gbko:v1";
+ static final String DIRECT_GABW_URN = "urn:beam:directrunner:transforms:gabw:v1";
+
DirectGroupByKey(GroupByKey<K, V> from) {
this.original = from;
}
@@ -68,7 +73,8 @@ class DirectGroupByKey<K, V>
}
static final class DirectGroupByKeyOnly<K, V>
- extends PTransform<PCollection<KV<K, V>>, PCollection<KeyedWorkItem<K, V>>> {
+ extends PTransformTranslation.RawPTransform<
+ PCollection<KV<K, V>>, PCollection<KeyedWorkItem<K, V>>, Message> {
@Override
public PCollection<KeyedWorkItem<K, V>> expand(PCollection<KV<K, V>> input) {
return PCollection.createPrimitiveOutputInternal(
@@ -86,10 +92,16 @@ class DirectGroupByKey<K, V>
GroupByKey.getInputValueCoder(input.getCoder()),
input.getWindowingStrategy().getWindowFn().windowCoder());
}
+
+ @Override
+ public String getUrn() {
+ return DIRECT_GBKO_URN;
+ }
}
static final class DirectGroupAlsoByWindow<K, V>
- extends PTransform<PCollection<KeyedWorkItem<K, V>>, PCollection<KV<K, Iterable<V>>>> {
+ extends PTransformTranslation.RawPTransform<
+ PCollection<KeyedWorkItem<K, V>>, PCollection<KV<K, Iterable<V>>>, Message> {
private final WindowingStrategy<?, ?> inputWindowingStrategy;
private final WindowingStrategy<?, ?> outputWindowingStrategy;
@@ -135,5 +147,10 @@ class DirectGroupByKey<K, V>
return PCollection.createPrimitiveOutputInternal(
input.getPipeline(), outputWindowingStrategy, input.isBounded());
}
+
+ @Override
+ public String getUrn() {
+ return DIRECT_GABW_URN;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/0e29cc52/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EmptyInputProvider.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EmptyInputProvider.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EmptyInputProvider.java
index c36879a..a5a53bc 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EmptyInputProvider.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/EmptyInputProvider.java
@@ -20,13 +20,12 @@ package org.apache.beam.runners.direct;
import java.util.Collection;
import java.util.Collections;
import org.apache.beam.sdk.runners.AppliedPTransform;
-import org.apache.beam.sdk.transforms.Flatten;
+import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionList;
/** A {@link RootInputProvider} that provides a singleton empty bundle. */
-class EmptyInputProvider<T>
- implements RootInputProvider<T, Void, PCollectionList<T>, Flatten.PCollections<T>> {
+class EmptyInputProvider<T> implements RootInputProvider<T, Void, PCollectionList<T>> {
EmptyInputProvider() {}
/**
@@ -36,7 +35,8 @@ class EmptyInputProvider<T>
*/
@Override
public Collection<CommittedBundle<Void>> getInitialInputs(
- AppliedPTransform<PCollectionList<T>, PCollection<T>, Flatten.PCollections<T>>
+ AppliedPTransform<
+ PCollectionList<T>, PCollection<T>, PTransform<PCollectionList<T>, PCollection<T>>>
transform,
int targetParallelism) {
return Collections.emptyList();
http://git-wip-us.apache.org/repos/asf/beam/blob/0e29cc52/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 be433db..df2054b 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,11 +19,13 @@ package org.apache.beam.runners.direct;
import static com.google.common.base.Preconditions.checkState;
+import com.google.protobuf.Message;
import java.util.Map;
import org.apache.beam.runners.core.KeyedWorkItem;
import org.apache.beam.runners.core.KeyedWorkItemCoder;
import org.apache.beam.runners.core.KeyedWorkItems;
import org.apache.beam.runners.core.construction.PTransformReplacements;
+import org.apache.beam.runners.core.construction.PTransformTranslation;
import org.apache.beam.runners.core.construction.ReplacementOutputs;
import org.apache.beam.runners.core.construction.SplittableParDo;
import org.apache.beam.sdk.coders.CannotProvideCoderException;
@@ -165,8 +167,12 @@ class ParDoMultiOverrideFactory<InputT, OutputT>
}
}
+ static final String DIRECT_STATEFUL_PAR_DO_URN =
+ "urn:beam:directrunner:transforms:stateful_pardo:v1";
+
static class StatefulParDo<K, InputT, OutputT>
- extends PTransform<PCollection<? extends KeyedWorkItem<K, KV<K, InputT>>>, PCollectionTuple> {
+ extends PTransformTranslation.RawPTransform<
+ PCollection<? extends KeyedWorkItem<K, KV<K, InputT>>>, PCollectionTuple, Message> {
private final transient MultiOutput<KV<K, InputT>, OutputT> underlyingParDo;
private final transient PCollection<KV<K, InputT>> originalInput;
@@ -201,6 +207,11 @@ class ParDoMultiOverrideFactory<InputT, OutputT>
return outputs;
}
+
+ @Override
+ public String getUrn() {
+ return DIRECT_STATEFUL_PAR_DO_URN;
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/beam/blob/0e29cc52/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ReadEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ReadEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ReadEvaluatorFactory.java
new file mode 100644
index 0000000..8521706
--- /dev/null
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ReadEvaluatorFactory.java
@@ -0,0 +1,97 @@
+/*
+ * 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.direct;
+
+import java.util.Collection;
+import javax.annotation.Nullable;
+import org.apache.beam.runners.core.construction.ReadTranslation;
+import org.apache.beam.sdk.io.Read;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.values.PBegin;
+import org.apache.beam.sdk.values.PCollection;
+
+/**
+ * A {@link TransformEvaluatorFactory} that produces {@link TransformEvaluator TransformEvaluators}
+ * for the {@link Read Read} primitives, whether bounded or unbounded.
+ */
+final class ReadEvaluatorFactory implements TransformEvaluatorFactory {
+
+ final BoundedReadEvaluatorFactory boundedFactory;
+ final UnboundedReadEvaluatorFactory unboundedFactory;
+
+ public ReadEvaluatorFactory(EvaluationContext context) {
+ boundedFactory = new BoundedReadEvaluatorFactory(context);
+ unboundedFactory = new UnboundedReadEvaluatorFactory(context);
+ }
+
+ @Nullable
+ @Override
+ public <InputT> TransformEvaluator<InputT> forApplication(
+ AppliedPTransform<?, ?, ?> application, CommittedBundle<?> inputBundle) throws Exception {
+ switch (ReadTranslation.sourceIsBounded(application)) {
+ case BOUNDED:
+ return boundedFactory.forApplication(application, inputBundle);
+ case UNBOUNDED:
+ return unboundedFactory.forApplication(application, inputBundle);
+ default:
+ throw new IllegalArgumentException("PCollection is neither bounded nor unbounded?!?");
+ }
+ }
+
+ @Override
+ public void cleanup() throws Exception {
+ boundedFactory.cleanup();
+ unboundedFactory.cleanup();
+ }
+
+ static <T> InputProvider<T> inputProvider(EvaluationContext context) {
+ return new InputProvider(context);
+ }
+
+ private static class InputProvider<T> implements RootInputProvider<T, SourceShard<T>, PBegin> {
+
+ private final UnboundedReadEvaluatorFactory.InputProvider<T> unboundedInputProvider;
+ private final BoundedReadEvaluatorFactory.InputProvider<T> boundedInputProvider;
+
+ InputProvider(EvaluationContext context) {
+ this.unboundedInputProvider = new UnboundedReadEvaluatorFactory.InputProvider<T>(context);
+ this.boundedInputProvider = new BoundedReadEvaluatorFactory.InputProvider<T>(context);
+ }
+
+ @Override
+ public Collection<CommittedBundle<SourceShard<T>>> getInitialInputs(
+ AppliedPTransform<PBegin, PCollection<T>, PTransform<PBegin, PCollection<T>>>
+ appliedTransform,
+ int targetParallelism)
+ throws Exception {
+ switch (ReadTranslation.sourceIsBounded(appliedTransform)) {
+ case BOUNDED:
+ // This cast could be made unnecessary, but too much bounded polymorphism
+ return (Collection)
+ boundedInputProvider.getInitialInputs(appliedTransform, targetParallelism);
+ case UNBOUNDED:
+ // This cast could be made unnecessary, but too much bounded polymorphism
+ return (Collection)
+ unboundedInputProvider.getInitialInputs(appliedTransform, targetParallelism);
+ default:
+ throw new IllegalArgumentException("PCollection is neither bounded nor unbounded?!?");
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/0e29cc52/runners/direct-java/src/main/java/org/apache/beam/runners/direct/RootInputProvider.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/RootInputProvider.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/RootInputProvider.java
index ce69518..0b3de32 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/RootInputProvider.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/RootInputProvider.java
@@ -29,8 +29,7 @@ import org.apache.beam.sdk.values.PInput;
* Provides {@link CommittedBundle bundles} that will be provided to the {@link PTransform
* PTransforms} that are at the root of a {@link Pipeline}.
*/
-interface RootInputProvider<
- T, ShardT, InputT extends PInput, TransformT extends PTransform<InputT, PCollection<T>>> {
+interface RootInputProvider<T, ShardT, InputT extends PInput> {
/**
* Get the initial inputs for the {@link AppliedPTransform}. The {@link AppliedPTransform} will be
* provided with these {@link CommittedBundle bundles} as input when the {@link Pipeline} runs.
@@ -44,6 +43,8 @@ interface RootInputProvider<
* greater than or equal to 1.
*/
Collection<CommittedBundle<ShardT>> getInitialInputs(
- AppliedPTransform<InputT, PCollection<T>, TransformT> transform, int targetParallelism)
+ AppliedPTransform<InputT, PCollection<T>, PTransform<InputT, PCollection<T>>>
+ transform,
+ int targetParallelism)
throws Exception;
}
http://git-wip-us.apache.org/repos/asf/beam/blob/0e29cc52/runners/direct-java/src/main/java/org/apache/beam/runners/direct/RootProviderRegistry.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/RootProviderRegistry.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/RootProviderRegistry.java
index 4b0c06d..5cbeab7 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/RootProviderRegistry.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/RootProviderRegistry.java
@@ -18,13 +18,14 @@
package org.apache.beam.runners.direct;
import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.beam.runners.core.construction.PTransformTranslation.FLATTEN_TRANSFORM_URN;
+import static org.apache.beam.runners.direct.TestStreamEvaluatorFactory.DirectTestStreamFactory.DIRECT_TEST_STREAM_URN;
import com.google.common.collect.ImmutableMap;
import java.util.Collection;
import java.util.Map;
-import org.apache.beam.sdk.io.Read;
+import org.apache.beam.runners.core.construction.PTransformTranslation;
import org.apache.beam.sdk.runners.AppliedPTransform;
-import org.apache.beam.sdk.transforms.Flatten.PCollections;
import org.apache.beam.sdk.transforms.PTransform;
/**
@@ -33,34 +34,31 @@ import org.apache.beam.sdk.transforms.PTransform;
*/
class RootProviderRegistry {
public static RootProviderRegistry defaultRegistry(EvaluationContext context) {
- ImmutableMap.Builder<Class<? extends PTransform>, RootInputProvider<?, ?, ?, ?>>
+ ImmutableMap.Builder<String, RootInputProvider<?, ?, ?>>
defaultProviders = ImmutableMap.builder();
defaultProviders
- .put(Read.Bounded.class, new BoundedReadEvaluatorFactory.InputProvider(context))
- .put(Read.Unbounded.class, new UnboundedReadEvaluatorFactory.InputProvider(context))
- .put(
- TestStreamEvaluatorFactory.DirectTestStreamFactory.DirectTestStream.class,
- new TestStreamEvaluatorFactory.InputProvider(context))
- .put(PCollections.class, new EmptyInputProvider());
+ .put(PTransformTranslation.READ_TRANSFORM_URN, ReadEvaluatorFactory.inputProvider(context))
+ .put(DIRECT_TEST_STREAM_URN, new TestStreamEvaluatorFactory.InputProvider(context))
+ .put(FLATTEN_TRANSFORM_URN, new EmptyInputProvider());
return new RootProviderRegistry(defaultProviders.build());
}
- private final Map<Class<? extends PTransform>, RootInputProvider<?, ?, ?, ?>> providers;
+ private final Map<String, RootInputProvider<?, ?, ?>> providers;
private RootProviderRegistry(
- Map<Class<? extends PTransform>, RootInputProvider<?, ?, ?, ?>> providers) {
+ Map<String, RootInputProvider<?, ?, ?>> providers) {
this.providers = providers;
}
public Collection<CommittedBundle<?>> getInitialInputs(
AppliedPTransform<?, ?, ?> transform, int targetParallelism) throws Exception {
- Class<? extends PTransform> transformClass = transform.getTransform().getClass();
+ String transformUrn = PTransformTranslation.urnForTransform(transform.getTransform());
RootInputProvider provider =
checkNotNull(
- providers.get(transformClass),
- "Tried to get a %s for a Transform of type %s, but there is no such provider",
+ providers.get(transformUrn),
+ "Tried to get a %s for a transform \"%s\", but there is no such provider",
RootInputProvider.class.getSimpleName(),
- transformClass.getSimpleName());
+ transformUrn);
return provider.getInitialInputs(transform, targetParallelism);
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/0e29cc52/runners/direct-java/src/main/java/org/apache/beam/runners/direct/SourceShard.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/SourceShard.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/SourceShard.java
new file mode 100644
index 0000000..a054333
--- /dev/null
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/SourceShard.java
@@ -0,0 +1,33 @@
+/*
+ * 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.direct;
+
+import org.apache.beam.sdk.io.BoundedSource;
+import org.apache.beam.sdk.io.Read;
+import org.apache.beam.sdk.io.Source;
+import org.apache.beam.sdk.io.UnboundedSource;
+
+/**
+ * A shard for a source in the {@link Read} transform.
+ *
+ * <p>Since {@link UnboundedSource} and {@link BoundedSource} have radically different needs, this
+ * is a mostly-empty interface.
+ */
+interface SourceShard<T> {
+ Source<T> getSource();
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/0e29cc52/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TestStreamEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TestStreamEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TestStreamEvaluatorFactory.java
index 8b21d5a..b1db58f 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TestStreamEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TestStreamEvaluatorFactory.java
@@ -22,12 +22,14 @@ import com.google.auto.value.AutoValue;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Supplier;
import com.google.common.collect.Iterables;
+import com.google.protobuf.Message;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import javax.annotation.Nullable;
+import org.apache.beam.runners.core.construction.PTransformTranslation;
import org.apache.beam.runners.core.construction.ReplacementOutputs;
import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.runners.PTransformOverrideFactory;
@@ -180,7 +182,10 @@ class TestStreamEvaluatorFactory implements TransformEvaluatorFactory {
return ReplacementOutputs.singleton(outputs, newOutput);
}
- static class DirectTestStream<T> extends PTransform<PBegin, PCollection<T>> {
+ static final String DIRECT_TEST_STREAM_URN = "urn:beam:directrunner:transforms:test_stream:v1";
+
+ static class DirectTestStream<T>
+ extends PTransformTranslation.RawPTransform<PBegin, PCollection<T>, Message> {
private final transient DirectRunner runner;
private final TestStream<T> original;
@@ -197,12 +202,15 @@ class TestStreamEvaluatorFactory implements TransformEvaluatorFactory {
input.getPipeline(), WindowingStrategy.globalDefault(), IsBounded.UNBOUNDED)
.setCoder(original.getValueCoder());
}
+
+ @Override
+ public String getUrn() {
+ return DIRECT_TEST_STREAM_URN;
+ }
}
}
- static class InputProvider<T>
- implements RootInputProvider<
- T, TestStreamIndex<T>, PBegin, DirectTestStreamFactory.DirectTestStream<T>> {
+ static class InputProvider<T> implements RootInputProvider<T, TestStreamIndex<T>, PBegin> {
private final EvaluationContext evaluationContext;
InputProvider(EvaluationContext evaluationContext) {
@@ -211,15 +219,17 @@ class TestStreamEvaluatorFactory implements TransformEvaluatorFactory {
@Override
public Collection<CommittedBundle<TestStreamIndex<T>>> getInitialInputs(
- AppliedPTransform<PBegin, PCollection<T>, DirectTestStreamFactory.DirectTestStream<T>>
- transform,
+ AppliedPTransform<PBegin, PCollection<T>, PTransform<PBegin, PCollection<T>>> transform,
int targetParallelism) {
+
+ // This will always be run on an execution-time transform, so it can be downcast
+ DirectTestStreamFactory.DirectTestStream<T> testStream =
+ (DirectTestStreamFactory.DirectTestStream<T>) transform.getTransform();
+
CommittedBundle<TestStreamIndex<T>> initialBundle =
evaluationContext
.<TestStreamIndex<T>>createRootBundle()
- .add(
- WindowedValue.valueInGlobalWindow(
- TestStreamIndex.of(transform.getTransform().original)))
+ .add(WindowedValue.valueInGlobalWindow(TestStreamIndex.of(testStream.original)))
.commit(BoundedWindow.TIMESTAMP_MAX_VALUE);
return Collections.singleton(initialBundle);
}
http://git-wip-us.apache.org/repos/asf/beam/blob/0e29cc52/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformEvaluatorRegistry.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformEvaluatorRegistry.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformEvaluatorRegistry.java
index 718cca2..d144b20 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformEvaluatorRegistry.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/TransformEvaluatorRegistry.java
@@ -19,23 +19,33 @@ package org.apache.beam.runners.direct;
import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
+import static org.apache.beam.runners.core.construction.PTransformTranslation.FLATTEN_TRANSFORM_URN;
+import static org.apache.beam.runners.core.construction.PTransformTranslation.PAR_DO_TRANSFORM_URN;
+import static org.apache.beam.runners.core.construction.PTransformTranslation.READ_TRANSFORM_URN;
+import static org.apache.beam.runners.core.construction.PTransformTranslation.WINDOW_TRANSFORM_URN;
+import static org.apache.beam.runners.core.construction.SplittableParDo.SPLITTABLE_PROCESS_URN;
+import static org.apache.beam.runners.direct.DirectGroupByKey.DIRECT_GABW_URN;
+import static org.apache.beam.runners.direct.DirectGroupByKey.DIRECT_GBKO_URN;
+import static org.apache.beam.runners.direct.ParDoMultiOverrideFactory.DIRECT_STATEFUL_PAR_DO_URN;
+import static org.apache.beam.runners.direct.TestStreamEvaluatorFactory.DirectTestStreamFactory.DIRECT_TEST_STREAM_URN;
+import static org.apache.beam.runners.direct.ViewOverrideFactory.DIRECT_WRITE_VIEW_URN;
+import com.google.auto.service.AutoService;
import com.google.common.collect.ImmutableMap;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.beam.runners.core.SplittableParDoViaKeyedWorkItems;
-import org.apache.beam.runners.direct.DirectGroupByKey.DirectGroupAlsoByWindow;
-import org.apache.beam.runners.direct.DirectGroupByKey.DirectGroupByKeyOnly;
-import org.apache.beam.runners.direct.ParDoMultiOverrideFactory.StatefulParDo;
-import org.apache.beam.runners.direct.ViewOverrideFactory.WriteView;
-import org.apache.beam.sdk.io.Read;
+import org.apache.beam.runners.core.SplittableParDoViaKeyedWorkItems.ProcessElements;
+import org.apache.beam.runners.core.construction.PTransformTranslation;
+import org.apache.beam.runners.core.construction.PTransformTranslation.TransformPayloadTranslator;
+import org.apache.beam.runners.core.construction.SdkComponents;
+import org.apache.beam.runners.core.construction.TransformPayloadTranslatorRegistrar;
+import org.apache.beam.runners.direct.TestStreamEvaluatorFactory.DirectTestStreamFactory.DirectTestStream;
+import org.apache.beam.sdk.common.runner.v1.RunnerApi.FunctionSpec;
import org.apache.beam.sdk.runners.AppliedPTransform;
-import org.apache.beam.sdk.transforms.Flatten.PCollections;
import org.apache.beam.sdk.transforms.PTransform;
-import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.transforms.windowing.Window;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -45,43 +55,93 @@ import org.slf4j.LoggerFactory;
*/
class TransformEvaluatorRegistry implements TransformEvaluatorFactory {
private static final Logger LOG = LoggerFactory.getLogger(TransformEvaluatorRegistry.class);
+
public static TransformEvaluatorRegistry defaultRegistry(EvaluationContext ctxt) {
- @SuppressWarnings({"rawtypes"})
- ImmutableMap<Class<? extends PTransform>, TransformEvaluatorFactory> primitives =
- ImmutableMap.<Class<? extends PTransform>, TransformEvaluatorFactory>builder()
- .put(Read.Bounded.class, new BoundedReadEvaluatorFactory(ctxt))
- .put(Read.Unbounded.class, new UnboundedReadEvaluatorFactory(ctxt))
+ ImmutableMap<String, TransformEvaluatorFactory> primitives =
+ ImmutableMap.<String, TransformEvaluatorFactory>builder()
+ // Beam primitives
+ .put(READ_TRANSFORM_URN, new ReadEvaluatorFactory(ctxt))
.put(
- ParDo.MultiOutput.class,
+ PAR_DO_TRANSFORM_URN,
new ParDoEvaluatorFactory<>(ctxt, ParDoEvaluator.defaultRunnerFactory()))
- .put(StatefulParDo.class, new StatefulParDoEvaluatorFactory<>(ctxt))
- .put(PCollections.class, new FlattenEvaluatorFactory(ctxt))
- .put(WriteView.class, new ViewEvaluatorFactory(ctxt))
- .put(Window.Assign.class, new WindowEvaluatorFactory(ctxt))
- // Runner-specific primitives used in expansion of GroupByKey
- .put(DirectGroupByKeyOnly.class, new GroupByKeyOnlyEvaluatorFactory(ctxt))
- .put(DirectGroupAlsoByWindow.class, new GroupAlsoByWindowEvaluatorFactory(ctxt))
- .put(
- TestStreamEvaluatorFactory.DirectTestStreamFactory.DirectTestStream.class,
- new TestStreamEvaluatorFactory(ctxt))
- // Runner-specific primitive used in expansion of SplittableParDo
- .put(
- SplittableParDoViaKeyedWorkItems.ProcessElements.class,
- new SplittableProcessElementsEvaluatorFactory<>(ctxt))
+ .put(FLATTEN_TRANSFORM_URN, new FlattenEvaluatorFactory(ctxt))
+ .put(WINDOW_TRANSFORM_URN, new WindowEvaluatorFactory(ctxt))
+
+ // Runner-specific primitives
+ .put(DIRECT_WRITE_VIEW_URN, new ViewEvaluatorFactory(ctxt))
+ .put(DIRECT_STATEFUL_PAR_DO_URN, new StatefulParDoEvaluatorFactory<>(ctxt))
+ .put(DIRECT_GBKO_URN, new GroupByKeyOnlyEvaluatorFactory(ctxt))
+ .put(DIRECT_GABW_URN, new GroupAlsoByWindowEvaluatorFactory(ctxt))
+ .put(DIRECT_TEST_STREAM_URN, new TestStreamEvaluatorFactory(ctxt))
+
+ // Runners-core primitives
+ .put(SPLITTABLE_PROCESS_URN, new SplittableProcessElementsEvaluatorFactory<>(ctxt))
.build();
return new TransformEvaluatorRegistry(primitives);
}
+ /** Registers classes specialized to the direct runner. */
+ @AutoService(TransformPayloadTranslatorRegistrar.class)
+ public static class DirectTransformsRegistrar implements TransformPayloadTranslatorRegistrar {
+ @Override
+ public Map<
+ ? extends Class<? extends PTransform>,
+ ? extends PTransformTranslation.TransformPayloadTranslator>
+ getTransformPayloadTranslators() {
+ return ImmutableMap
+ .<Class<? extends PTransform>, PTransformTranslation.TransformPayloadTranslator>builder()
+ .put(
+ DirectGroupByKey.DirectGroupByKeyOnly.class,
+ new PTransformTranslation.RawPTransformTranslator<>())
+ .put(
+ DirectGroupByKey.DirectGroupAlsoByWindow.class,
+ new PTransformTranslation.RawPTransformTranslator())
+ .put(
+ ParDoMultiOverrideFactory.StatefulParDo.class,
+ new PTransformTranslation.RawPTransformTranslator<>())
+ .put(
+ ViewOverrideFactory.WriteView.class,
+ new PTransformTranslation.RawPTransformTranslator<>())
+ .put(DirectTestStream.class, new PTransformTranslation.RawPTransformTranslator<>())
+ .put(
+ SplittableParDoViaKeyedWorkItems.ProcessElements.class,
+ new SplittableParDoProcessElementsTranslator())
+ .build();
+ }
+ }
+
+ /**
+ * A translator just to vend the URN. This will need to be moved to runners-core-construction-java
+ * once SDF is reorganized appropriately.
+ */
+ private static class SplittableParDoProcessElementsTranslator
+ implements TransformPayloadTranslator<ProcessElements<?, ?, ?, ?>> {
+
+ private SplittableParDoProcessElementsTranslator() {}
+
+ @Override
+ public String getUrn(ProcessElements<?, ?, ?, ?> transform) {
+ return SPLITTABLE_PROCESS_URN;
+ }
+
+ @Override
+ public FunctionSpec translate(
+ AppliedPTransform<?, ?, ProcessElements<?, ?, ?, ?>> transform, SdkComponents components) {
+ throw new UnsupportedOperationException(
+ String.format("%s should never be translated",
+ ProcessElements.class.getCanonicalName()));
+ }
+ }
+
// the TransformEvaluatorFactories can construct instances of all generic types of transform,
// so all instances of a primitive can be handled with the same evaluator factory.
- @SuppressWarnings("rawtypes")
- private final Map<Class<? extends PTransform>, TransformEvaluatorFactory> factories;
+ private final Map<String, TransformEvaluatorFactory> factories;
private final AtomicBoolean finished = new AtomicBoolean(false);
private TransformEvaluatorRegistry(
@SuppressWarnings("rawtypes")
- Map<Class<? extends PTransform>, TransformEvaluatorFactory> factories) {
+ Map<String, TransformEvaluatorFactory> factories) {
this.factories = factories;
}
@@ -91,10 +151,12 @@ class TransformEvaluatorRegistry implements TransformEvaluatorFactory {
throws Exception {
checkState(
!finished.get(), "Tried to get an evaluator for a finished TransformEvaluatorRegistry");
- Class<? extends PTransform> transformClass = application.getTransform().getClass();
+
+ String urn = PTransformTranslation.urnForTransform(application.getTransform());
+
TransformEvaluatorFactory factory =
checkNotNull(
- factories.get(transformClass), "No evaluator for PTransform type %s", transformClass);
+ factories.get(urn), "No evaluator for PTransform \"%s\"", urn);
return factory.forApplication(application, inputBundle);
}
http://git-wip-us.apache.org/repos/asf/beam/blob/0e29cc52/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java
index cba826c..7d4bba1 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/UnboundedReadEvaluatorFactory.java
@@ -29,6 +29,7 @@ import java.util.Collections;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import javax.annotation.Nullable;
+import org.apache.beam.runners.core.construction.ReadTranslation;
import org.apache.beam.runners.direct.UnboundedReadDeduplicator.NeverDeduplicator;
import org.apache.beam.sdk.io.Read;
import org.apache.beam.sdk.io.Read.Unbounded;
@@ -253,7 +254,8 @@ class UnboundedReadEvaluatorFactory implements TransformEvaluatorFactory {
}
@AutoValue
- abstract static class UnboundedSourceShard<T, CheckpointT extends CheckpointMark> {
+ abstract static class UnboundedSourceShard<T, CheckpointT extends CheckpointMark>
+ implements SourceShard<T> {
static <T, CheckpointT extends CheckpointMark> UnboundedSourceShard<T, CheckpointT> unstarted(
UnboundedSource<T, CheckpointT> source, UnboundedReadDeduplicator deduplicator) {
return of(source, deduplicator, null, null);
@@ -268,7 +270,8 @@ class UnboundedReadEvaluatorFactory implements TransformEvaluatorFactory {
source, deduplicator, reader, checkpoint);
}
- abstract UnboundedSource<T, CheckpointT> getSource();
+ @Override
+ public abstract UnboundedSource<T, CheckpointT> getSource();
abstract UnboundedReadDeduplicator getDeduplicator();
@@ -283,9 +286,8 @@ class UnboundedReadEvaluatorFactory implements TransformEvaluatorFactory {
}
}
- static class InputProvider<OutputT>
- implements RootInputProvider<
- OutputT, UnboundedSourceShard<OutputT, ?>, PBegin, Unbounded<OutputT>> {
+ static class InputProvider<T>
+ implements RootInputProvider<T, UnboundedSourceShard<T, ?>, PBegin> {
private final EvaluationContext evaluationContext;
InputProvider(EvaluationContext evaluationContext) {
@@ -293,27 +295,28 @@ class UnboundedReadEvaluatorFactory implements TransformEvaluatorFactory {
}
@Override
- public Collection<CommittedBundle<UnboundedSourceShard<OutputT, ?>>> getInitialInputs(
- AppliedPTransform<PBegin, PCollection<OutputT>, Unbounded<OutputT>> transform,
+ public Collection<CommittedBundle<UnboundedSourceShard<T, ?>>> getInitialInputs(
+ AppliedPTransform<PBegin, PCollection<T>, PTransform<PBegin, PCollection<T>>>
+ transform,
int targetParallelism)
throws Exception {
- UnboundedSource<OutputT, ?> source = transform.getTransform().getSource();
- List<? extends UnboundedSource<OutputT, ?>> splits =
+ UnboundedSource<T, ?> source = ReadTranslation.unboundedSourceFromTransform(transform);
+ List<? extends UnboundedSource<T, ?>> splits =
source.split(targetParallelism, evaluationContext.getPipelineOptions());
UnboundedReadDeduplicator deduplicator =
source.requiresDeduping()
? UnboundedReadDeduplicator.CachedIdDeduplicator.create()
: NeverDeduplicator.create();
- ImmutableList.Builder<CommittedBundle<UnboundedSourceShard<OutputT, ?>>> initialShards =
+ ImmutableList.Builder<CommittedBundle<UnboundedSourceShard<T, ?>>> initialShards =
ImmutableList.builder();
- for (UnboundedSource<OutputT, ?> split : splits) {
- UnboundedSourceShard<OutputT, ?> shard =
+ for (UnboundedSource<T, ?> split : splits) {
+ UnboundedSourceShard<T, ?> shard =
UnboundedSourceShard.unstarted(split, deduplicator);
initialShards.add(
evaluationContext
- .<UnboundedSourceShard<OutputT, ?>>createRootBundle()
- .add(WindowedValue.<UnboundedSourceShard<OutputT, ?>>valueInGlobalWindow(shard))
+ .<UnboundedSourceShard<T, ?>>createRootBundle()
+ .add(WindowedValue.<UnboundedSourceShard<T, ?>>valueInGlobalWindow(shard))
.commit(BoundedWindow.TIMESTAMP_MAX_VALUE));
}
return initialShards.build();
http://git-wip-us.apache.org/repos/asf/beam/blob/0e29cc52/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ViewOverrideFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ViewOverrideFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ViewOverrideFactory.java
index b3bbac8..501b436 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ViewOverrideFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ViewOverrideFactory.java
@@ -18,10 +18,12 @@
package org.apache.beam.runners.direct;
+import com.google.protobuf.Message;
import java.util.Collections;
import java.util.Map;
import org.apache.beam.runners.core.construction.ForwardingPTransform;
import org.apache.beam.runners.core.construction.PTransformReplacements;
+import org.apache.beam.runners.core.construction.PTransformTranslation.RawPTransform;
import org.apache.beam.sdk.coders.KvCoder;
import org.apache.beam.sdk.coders.VoidCoder;
import org.apache.beam.sdk.runners.AppliedPTransform;
@@ -93,7 +95,7 @@ class ViewOverrideFactory<ElemT, ViewT>
* to {@link ViewT}.
*/
static final class WriteView<ElemT, ViewT>
- extends PTransform<PCollection<Iterable<ElemT>>, PCollectionView<ViewT>> {
+ extends RawPTransform<PCollection<Iterable<ElemT>>, PCollectionView<ViewT>, Message> {
private final CreatePCollectionView<ElemT, ViewT> og;
WriteView(CreatePCollectionView<ElemT, ViewT> og) {
@@ -110,5 +112,13 @@ class ViewOverrideFactory<ElemT, ViewT>
public PCollectionView<ViewT> getView() {
return og.getView();
}
+
+ @Override
+ public String getUrn() {
+ return DIRECT_WRITE_VIEW_URN;
+ }
}
+
+ public static final String DIRECT_WRITE_VIEW_URN =
+ "urn:beam:directrunner:transforms:write_view:v1";
}
http://git-wip-us.apache.org/repos/asf/beam/blob/0e29cc52/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml
----------------------------------------------------------------------
diff --git a/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml b/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml
index 8ff0cb0..3430750 100644
--- a/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml
+++ b/sdks/java/build-tools/src/main/resources/beam/findbugs-filter.xml
@@ -405,4 +405,11 @@
<Bug pattern="NM_CLASS_NOT_EXCEPTION"/>
<!-- It is clear from the name that this class holds either StorageObject or IOException. -->
</Match>
+
+ <Match>
+ <Class name="org.apache.beam.runners.direct.ParDoMultiOverrideFactory$StatefulParDo"/>
+ <Bug pattern="SE_TRANSIENT_FIELD_NOT_RESTORED"/>
+ <!-- PTransforms do not actually support serialization. -->
+ </Match>
+
</FindBugsFilter>
[02/12] beam git commit: Make SdkComponents public for
TransformPayloadTranslator
Posted by ke...@apache.org.
Make SdkComponents public for TransformPayloadTranslator
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/26c93681
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/26c93681
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/26c93681
Branch: refs/heads/master
Commit: 26c93681795dc9b98c3f00c2cd61c38b36656e0b
Parents: 3643676
Author: Kenneth Knowles <kl...@google.com>
Authored: Fri May 19 21:35:43 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu May 25 11:14:29 2017 -0700
----------------------------------------------------------------------
.../org/apache/beam/runners/core/construction/SdkComponents.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/26c93681/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/SdkComponents.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/SdkComponents.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/SdkComponents.java
index b0f164f..0d3ba60 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/SdkComponents.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/SdkComponents.java
@@ -46,7 +46,7 @@ import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.WindowingStrategy;
/** SDK objects that will be represented at some later point within a {@link Components} object. */
-class SdkComponents {
+public class SdkComponents {
private final RunnerApi.Components.Builder componentsBuilder;
private final BiMap<AppliedPTransform<?, ?, ?>, String> transformIds;
[08/12] beam git commit: Add registration for Read and WindowInto
translators
Posted by ke...@apache.org.
Add registration for Read and WindowInto translators
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/afaebb13
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/afaebb13
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/afaebb13
Branch: refs/heads/master
Commit: afaebb13f73868799b83c7af95cca2732e3ecb9a
Parents: 0bf4ddb
Author: Kenneth Knowles <kl...@google.com>
Authored: Mon May 22 21:00:02 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu May 25 11:16:47 2017 -0700
----------------------------------------------------------------------
.../core/construction/ReadTranslation.java | 74 +++++++++++++++++++-
.../construction/WindowIntoTranslation.java | 43 +++++++++++-
2 files changed, 115 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/afaebb13/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
index d6c3400..aff5fc9 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/ReadTranslation.java
@@ -20,10 +20,15 @@ package org.apache.beam.runners.core.construction;
import static com.google.common.base.Preconditions.checkArgument;
+import com.google.auto.service.AutoService;
+import com.google.common.collect.ImmutableMap;
import com.google.protobuf.Any;
import com.google.protobuf.ByteString;
import com.google.protobuf.BytesValue;
import com.google.protobuf.InvalidProtocolBufferException;
+import java.util.Map;
+import org.apache.beam.runners.core.construction.PTransformTranslation.TransformPayloadTranslator;
+import org.apache.beam.sdk.common.runner.v1.RunnerApi;
import org.apache.beam.sdk.common.runner.v1.RunnerApi.FunctionSpec;
import org.apache.beam.sdk.common.runner.v1.RunnerApi.IsBounded;
import org.apache.beam.sdk.common.runner.v1.RunnerApi.ReadPayload;
@@ -32,12 +37,13 @@ import org.apache.beam.sdk.io.BoundedSource;
import org.apache.beam.sdk.io.Read;
import org.apache.beam.sdk.io.Source;
import org.apache.beam.sdk.io.UnboundedSource;
+import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.util.SerializableUtils;
/**
* Methods for translating {@link Read.Bounded} and {@link Read.Unbounded}
- * {@link PTransform PTransforms} into {@link ReadPayload} protos.
+ * {@link PTransform PTransformTranslation} into {@link ReadPayload} protos.
*/
public class ReadTranslation {
private static final String JAVA_SERIALIZED_BOUNDED_SOURCE = "urn:beam:java:boundedsource:v1";
@@ -124,4 +130,70 @@ public class ReadTranslation {
"BoundedSource");
}
+ /**
+ * A {@link TransformPayloadTranslator} for {@link Read.Unbounded}.
+ */
+ public static class UnboundedReadPayloadTranslator
+ implements PTransformTranslation.TransformPayloadTranslator<Read.Unbounded<?>> {
+ public static TransformPayloadTranslator create() {
+ return new UnboundedReadPayloadTranslator();
+ }
+
+ private UnboundedReadPayloadTranslator() {}
+
+ @Override
+ public String getUrn(Read.Unbounded<?> transform) {
+ return PTransformTranslation.WINDOW_TRANSFORM_URN;
+ }
+
+ @Override
+ public FunctionSpec translate(
+ AppliedPTransform<?, ?, Read.Unbounded<?>> transform, SdkComponents components) {
+ ReadPayload payload = toProto(transform.getTransform());
+ return RunnerApi.FunctionSpec.newBuilder()
+ .setUrn(PTransformTranslation.READ_TRANSFORM_URN)
+ .setParameter(Any.pack(payload))
+ .build();
+ }
+ }
+
+ /**
+ * A {@link TransformPayloadTranslator} for {@link Read.Bounded}.
+ */
+ public static class BoundedReadPayloadTranslator
+ implements PTransformTranslation.TransformPayloadTranslator<Read.Bounded<?>> {
+ public static TransformPayloadTranslator create() {
+ return new BoundedReadPayloadTranslator();
+ }
+
+ private BoundedReadPayloadTranslator() {}
+
+ @Override
+ public String getUrn(Read.Bounded<?> transform) {
+ return PTransformTranslation.WINDOW_TRANSFORM_URN;
+ }
+
+ @Override
+ public FunctionSpec translate(
+ AppliedPTransform<?, ?, Read.Bounded<?>> transform, SdkComponents components) {
+ ReadPayload payload = toProto(transform.getTransform());
+ return RunnerApi.FunctionSpec.newBuilder()
+ .setUrn(PTransformTranslation.READ_TRANSFORM_URN)
+ .setParameter(Any.pack(payload))
+ .build();
+ }
+ }
+
+ /** Registers {@link UnboundedReadPayloadTranslator} and {@link BoundedReadPayloadTranslator}. */
+ @AutoService(TransformPayloadTranslatorRegistrar.class)
+ public static class Registrar implements TransformPayloadTranslatorRegistrar {
+ @Override
+ public Map<? extends Class<? extends PTransform>, ? extends TransformPayloadTranslator>
+ getTransformPayloadTranslators() {
+ return ImmutableMap.<Class<? extends PTransform>, TransformPayloadTranslator>builder()
+ .put(Read.Unbounded.class, new UnboundedReadPayloadTranslator())
+ .put(Read.Bounded.class, new BoundedReadPayloadTranslator())
+ .build();
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/beam/blob/afaebb13/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowIntoTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowIntoTranslation.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowIntoTranslation.java
index 33faa02..5ed4d24 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowIntoTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowIntoTranslation.java
@@ -18,14 +18,18 @@
package org.apache.beam.runners.core.construction;
+import com.google.auto.service.AutoService;
import com.google.protobuf.Any;
import com.google.protobuf.InvalidProtocolBufferException;
+import java.util.Collections;
+import java.util.Map;
import org.apache.beam.runners.core.construction.PTransformTranslation.TransformPayloadTranslator;
import org.apache.beam.sdk.common.runner.v1.RunnerApi;
import org.apache.beam.sdk.common.runner.v1.RunnerApi.FunctionSpec;
import org.apache.beam.sdk.common.runner.v1.RunnerApi.SdkFunctionSpec;
import org.apache.beam.sdk.common.runner.v1.RunnerApi.WindowIntoPayload;
import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.transforms.windowing.Window.Assign;
import org.apache.beam.sdk.transforms.windowing.WindowFn;
@@ -40,7 +44,7 @@ public class WindowIntoTranslation {
@Override
public String getUrn(Assign<?> transform) {
- return PTransforms.WINDOW_TRANSFORM_URN;
+ return PTransformTranslation.WINDOW_TRANSFORM_URN;
}
@Override
@@ -65,4 +69,41 @@ public class WindowIntoTranslation {
SdkFunctionSpec spec = payload.getWindowFn();
return WindowingStrategyTranslation.windowFnFromProto(spec);
}
+
+ /**
+ * A {@link TransformPayloadTranslator} for {@link Window}.
+ */
+ public static class WindowIntoPayloadTranslator
+ implements PTransformTranslation.TransformPayloadTranslator<Window.Assign<?>> {
+ public static TransformPayloadTranslator create() {
+ return new WindowIntoPayloadTranslator();
+ }
+
+ private WindowIntoPayloadTranslator() {}
+
+ @Override
+ public String getUrn(Window.Assign<?> transform) {
+ return PTransformTranslation.WINDOW_TRANSFORM_URN;
+ }
+
+ @Override
+ public FunctionSpec translate(
+ AppliedPTransform<?, ?, Window.Assign<?>> transform, SdkComponents components) {
+ WindowIntoPayload payload = toProto(transform.getTransform(), components);
+ return RunnerApi.FunctionSpec.newBuilder()
+ .setUrn(PTransformTranslation.WINDOW_TRANSFORM_URN)
+ .setParameter(Any.pack(payload))
+ .build();
+ }
+ }
+
+ /** Registers {@link WindowIntoPayloadTranslator}. */
+ @AutoService(TransformPayloadTranslatorRegistrar.class)
+ public static class Registrar implements TransformPayloadTranslatorRegistrar {
+ @Override
+ public Map<? extends Class<? extends PTransform>, ? extends TransformPayloadTranslator>
+ getTransformPayloadTranslators() {
+ return Collections.singletonMap(Window.Assign.class, new WindowIntoPayloadTranslator());
+ }
+ }
}
[07/12] beam git commit: Add trivial FlattenTranslator to access URN
Posted by ke...@apache.org.
Add trivial FlattenTranslator to access URN
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/121631c6
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/121631c6
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/121631c6
Branch: refs/heads/master
Commit: 121631c62f94a6e553d3bbd8708cee2dbdf53923
Parents: 69bb1be
Author: Kenneth Knowles <kl...@google.com>
Authored: Mon May 22 21:09:54 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu May 25 11:16:47 2017 -0700
----------------------------------------------------------------------
.../core/construction/FlattenTranslator.java | 63 ++++++++++++++++++++
1 file changed, 63 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/121631c6/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/FlattenTranslator.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/FlattenTranslator.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/FlattenTranslator.java
new file mode 100644
index 0000000..f1d553d
--- /dev/null
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/FlattenTranslator.java
@@ -0,0 +1,63 @@
+/*
+ * 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.core.construction;
+
+import com.google.auto.service.AutoService;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.beam.runners.core.construction.PTransformTranslation.TransformPayloadTranslator;
+import org.apache.beam.sdk.common.runner.v1.RunnerApi;
+import org.apache.beam.sdk.common.runner.v1.RunnerApi.FunctionSpec;
+import org.apache.beam.sdk.runners.AppliedPTransform;
+import org.apache.beam.sdk.transforms.Flatten;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.windowing.Window.Assign;
+
+/**
+ * Utility methods for translating a {@link Assign} to and from {@link RunnerApi} representations.
+ */
+public class FlattenTranslator implements TransformPayloadTranslator<Flatten.PCollections<?>> {
+
+ public static TransformPayloadTranslator create() {
+ return new FlattenTranslator();
+ }
+
+ private FlattenTranslator() {}
+
+ @Override
+ public String getUrn(Flatten.PCollections<?> transform) {
+ return PTransformTranslation.FLATTEN_TRANSFORM_URN;
+ }
+
+ @Override
+ public FunctionSpec translate(
+ AppliedPTransform<?, ?, Flatten.PCollections<?>> transform, SdkComponents components) {
+ return RunnerApi.FunctionSpec.newBuilder().setUrn(getUrn(transform.getTransform())).build();
+ }
+
+ /** Registers {@link FlattenTranslator}. */
+ @AutoService(TransformPayloadTranslatorRegistrar.class)
+ public static class Registrar implements TransformPayloadTranslatorRegistrar {
+ @Override
+ public Map<? extends Class<? extends PTransform>, ? extends TransformPayloadTranslator>
+ getTransformPayloadTranslators() {
+ return Collections.singletonMap(Flatten.PCollections.class, new FlattenTranslator());
+ }
+ }
+}
[06/12] beam git commit: Register WindowIntoTranslator
Posted by ke...@apache.org.
Register WindowIntoTranslator
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/69bb1be4
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/69bb1be4
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/69bb1be4
Branch: refs/heads/master
Commit: 69bb1be4d3bd20e5c080bcce0eb464fac3d2ff73
Parents: d91c840
Author: Kenneth Knowles <kl...@google.com>
Authored: Mon May 22 21:09:33 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu May 25 11:16:47 2017 -0700
----------------------------------------------------------------------
.../beam/runners/core/construction/WindowIntoTranslation.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/69bb1be4/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowIntoTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowIntoTranslation.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowIntoTranslation.java
index 5ed4d24..aa17bc9 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowIntoTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/WindowIntoTranslation.java
@@ -91,7 +91,7 @@ public class WindowIntoTranslation {
AppliedPTransform<?, ?, Window.Assign<?>> transform, SdkComponents components) {
WindowIntoPayload payload = toProto(transform.getTransform(), components);
return RunnerApi.FunctionSpec.newBuilder()
- .setUrn(PTransformTranslation.WINDOW_TRANSFORM_URN)
+ .setUrn(getUrn(transform.getTransform()))
.setParameter(Any.pack(payload))
.build();
}
[11/12] beam git commit: Add RawPTransform,
which can just vend its URN and payload
Posted by ke...@apache.org.
Add RawPTransform, which can just vend its URN and payload
This is the type that will be returned when a pipeline is deserialized.
This also is convenient for direct runner overrides which do not really
merit translator registrations, yet URNs need to be known in order to
key the evaluator registry off URN.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/663ad881
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/663ad881
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/663ad881
Branch: refs/heads/master
Commit: 663ad88178ceadffe4cfa592555986ed7dde58b4
Parents: 8e09596
Author: Kenneth Knowles <kl...@google.com>
Authored: Fri May 19 21:28:19 2017 -0700
Committer: Kenneth Knowles <kl...@google.com>
Committed: Thu May 25 11:17:08 2017 -0700
----------------------------------------------------------------------
.../construction/PTransformTranslation.java | 51 ++++++++++++++++++++
1 file changed, 51 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/663ad881/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java
----------------------------------------------------------------------
diff --git a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java
index 35bb0e3..9f5f3b5 100644
--- a/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java
+++ b/runners/core-construction-java/src/main/java/org/apache/beam/runners/core/construction/PTransformTranslation.java
@@ -21,15 +21,20 @@ package org.apache.beam.runners.core.construction;
import static com.google.common.base.Preconditions.checkArgument;
import com.google.common.collect.ImmutableMap;
+import com.google.protobuf.Any;
+import com.google.protobuf.Message;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.ServiceLoader;
+import javax.annotation.Nullable;
import org.apache.beam.sdk.common.runner.v1.RunnerApi;
import org.apache.beam.sdk.common.runner.v1.RunnerApi.FunctionSpec;
import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PInput;
+import org.apache.beam.sdk.values.POutput;
import org.apache.beam.sdk.values.PValue;
import org.apache.beam.sdk.values.TupleTag;
@@ -135,4 +140,50 @@ public class PTransformTranslation {
String getUrn(T transform);
FunctionSpec translate(AppliedPTransform<?, ?, T> application, SdkComponents components);
}
+
+ /**
+ * A {@link PTransform} that indicates its URN and payload directly.
+ *
+ * <p>This is the result of rehydrating transforms from a pipeline proto. There is no {@link
+ * #expand} method since the definition of the transform may be lost. The transform is already
+ * fully expanded in the pipeline proto.
+ */
+ public abstract static class RawPTransform<
+ InputT extends PInput, OutputT extends POutput, PayloadT extends Message>
+ extends PTransform<InputT, OutputT> {
+
+ public abstract String getUrn();
+
+ @Nullable
+ PayloadT getPayload() {
+ return null;
+ }
+ }
+
+ /**
+ * A translator that uses the explicit URN and payload from a {@link RawPTransform}.
+ */
+ public static class RawPTransformTranslator<PayloadT extends Message>
+ implements TransformPayloadTranslator<RawPTransform<?, ?, PayloadT>> {
+ @Override
+ public String getUrn(RawPTransform<?, ?, PayloadT> transform) {
+ return transform.getUrn();
+ }
+
+ @Override
+ public FunctionSpec translate(
+ AppliedPTransform<?, ?, RawPTransform<?, ?, PayloadT>> transform,
+ SdkComponents components) {
+ PayloadT payload = transform.getTransform().getPayload();
+
+ FunctionSpec.Builder transformSpec =
+ FunctionSpec.newBuilder().setUrn(getUrn(transform.getTransform()));
+
+ if (payload != null) {
+ transformSpec.setParameter(Any.pack(payload));
+ }
+
+ return transformSpec.build();
+ }
+ }
}