beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aljos...@apache.org
Subject [1/2] incubator-beam git commit: [BEAM-965] Set Correct Output Type on Sources in Flink Stream Runner
Date Mon, 14 Nov 2016 21:08:47 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/master 8bfa08519 -> ee55f6e39


[BEAM-965] Set Correct Output Type on Sources in Flink Stream Runner


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/a61832a8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/a61832a8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/a61832a8

Branch: refs/heads/master
Commit: a61832a83c15e0aa5e3cc1985fc32aa1afb348e6
Parents: 8bfa085
Author: Aljoscha Krettek <aljoscha.krettek@gmail.com>
Authored: Fri Nov 11 23:58:16 2016 +0100
Committer: Aljoscha Krettek <aljoscha.krettek@gmail.com>
Committed: Mon Nov 14 22:07:45 2016 +0100

----------------------------------------------------------------------
 .../FlinkStreamingTransformTranslators.java            | 13 ++++++++++---
 1 file changed, 10 insertions(+), 3 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a61832a8/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/FlinkStreamingTransformTranslators.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/FlinkStreamingTransformTranslators.java
b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/FlinkStreamingTransformTranslators.java
index a3e8a49..687e9c8 100644
--- a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/FlinkStreamingTransformTranslators.java
+++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/FlinkStreamingTransformTranslators.java
@@ -219,6 +219,9 @@ public class FlinkStreamingTransformTranslators {
         FlinkStreamingTranslationContext context) {
       PCollection<T> output = context.getOutput(transform);
 
+      TypeInformation<WindowedValue<T>> outputTypeInfo =
+          context.getTypeInfo(context.getOutput(transform));
+
       DataStream<WindowedValue<T>> source;
       if (transform.getSource().getClass().equals(UnboundedFlinkSource.class)) {
         @SuppressWarnings("unchecked")
@@ -246,7 +249,7 @@ public class FlinkStreamingTransformTranslators {
                         new Instant(flinkAssigner.extractTimestamp(s, -1)),
                         GlobalWindow.INSTANCE,
                         PaneInfo.NO_FIRING));
-              }});
+              }}).returns(outputTypeInfo);
       } else {
         try {
           UnboundedSourceWrapper<T, ?> sourceWrapper =
@@ -256,7 +259,7 @@ public class FlinkStreamingTransformTranslators {
                   context.getExecutionEnvironment().getParallelism());
           source = context
               .getExecutionEnvironment()
-              .addSource(sourceWrapper).name(transform.getName());
+              .addSource(sourceWrapper).name(transform.getName()).returns(outputTypeInfo);
         } catch (Exception e) {
           throw new RuntimeException(
               "Error while translating UnboundedSource: " + transform.getSource(), e);
@@ -276,6 +279,10 @@ public class FlinkStreamingTransformTranslators {
         FlinkStreamingTranslationContext context) {
       PCollection<T> output = context.getOutput(transform);
 
+      TypeInformation<WindowedValue<T>> outputTypeInfo =
+          context.getTypeInfo(context.getOutput(transform));
+
+
       DataStream<WindowedValue<T>> source;
       try {
         BoundedSourceWrapper<T> sourceWrapper =
@@ -285,7 +292,7 @@ public class FlinkStreamingTransformTranslators {
                 context.getExecutionEnvironment().getParallelism());
         source = context
             .getExecutionEnvironment()
-            .addSource(sourceWrapper).name(transform.getName());
+            .addSource(sourceWrapper).name(transform.getName()).returns(outputTypeInfo);
       } catch (Exception e) {
         throw new RuntimeException(
             "Error while translating BoundedSource: " + transform.getSource(), e);


Mime
View raw message