beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dhalp...@apache.org
Subject [2/3] beam git commit: Add RawUnion code to FlinkDoFnFunction
Date Wed, 31 May 2017 18:36:25 GMT
Add RawUnion code to FlinkDoFnFunction


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

Branch: refs/heads/master
Commit: 5780fc5e8cd64911d9612d89896b9d68be4f621f
Parents: a0444b8
Author: Dan Halperin <dhalperi@google.com>
Authored: Wed May 31 10:50:46 2017 -0700
Committer: Dan Halperin <dhalperi@google.com>
Committed: Wed May 31 11:36:16 2017 -0700

----------------------------------------------------------------------
 .../runners/flink/translation/functions/FlinkDoFnFunction.java   | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/5780fc5e/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java
b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java
index ab2ac6b..d8ed622 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/functions/FlinkDoFnFunction.java
@@ -146,7 +146,9 @@ public class FlinkDoFnFunction<InputT, OutputT>
     @Override
     @SuppressWarnings("unchecked")
     public <T> void output(TupleTag<T> tag, WindowedValue<T> output) {
-      collector.collect(output);
+      collector.collect(
+          WindowedValue.of(new RawUnionValue(0 /* single output */, output.getValue()),
+          output.getTimestamp(), output.getWindows(), output.getPane()));
     }
   }
 


Mime
View raw message