beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From k...@apache.org
Subject [04/12] beam git commit: Allow getting URN for class of transform via translator
Date Fri, 26 May 2017 04:08:10 GMT
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 <klk@google.com>
Authored: Fri May 19 20:55:25 2017 -0700
Committer: Kenneth Knowles <klk@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)
{


Mime
View raw message