beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From m..@apache.org
Subject [2/4] incubator-beam git commit: fix Flink source coder handling
Date Fri, 29 Apr 2016 15:58:58 GMT
fix Flink source coder handling


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

Branch: refs/heads/master
Commit: aead96ff4c018b96a7b5ab1defb408c2a09b1be7
Parents: bc847a9
Author: Maximilian Michels <mxm@apache.org>
Authored: Thu Apr 28 12:00:18 2016 +0200
Committer: Maximilian Michels <mxm@apache.org>
Committed: Fri Apr 29 17:58:00 2016 +0200

----------------------------------------------------------------------
 .../FlinkStreamingTransformTranslators.java     | 13 +++-
 .../flink/translation/types/FlinkCoder.java     | 64 ++++++++++++++++++++
 .../streaming/io/UnboundedFlinkSource.java      | 12 +++-
 3 files changed, 84 insertions(+), 5 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/aead96ff/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 db24f9d..618727d 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
@@ -20,6 +20,7 @@ package org.apache.beam.runners.flink.translation;
 
 import org.apache.beam.runners.flink.translation.functions.UnionCoder;
 import org.apache.beam.runners.flink.translation.types.CoderTypeInformation;
+import org.apache.beam.runners.flink.translation.types.FlinkCoder;
 import org.apache.beam.runners.flink.translation.wrappers.SourceInputFormat;
 import org.apache.beam.runners.flink.translation.wrappers.streaming.FlinkGroupAlsoByWindowWrapper;
 import org.apache.beam.runners.flink.translation.wrappers.streaming.FlinkGroupByKeyWrapper;
@@ -262,9 +263,15 @@ public class FlinkStreamingTransformTranslators {
       DataStream<WindowedValue<T>> source;
       if (transform.getSource().getClass().equals(UnboundedFlinkSource.class)) {
         @SuppressWarnings("unchecked")
-        UnboundedFlinkSource<T> flinkSource = (UnboundedFlinkSource<T>) transform.getSource();
-        source = context.getExecutionEnvironment()
-            .addSource(flinkSource.getFlinkSource())
+        UnboundedFlinkSource<T> flinkSourceFunction = (UnboundedFlinkSource<T>)
transform.getSource();
+        DataStream<T> flinkSource = context.getExecutionEnvironment()
+            .addSource(flinkSourceFunction.getFlinkSource());
+
+        flinkSourceFunction.setCoder(
+            new FlinkCoder<T>(flinkSource.getType(),
+              context.getExecutionEnvironment().getConfig()));
+
+        source = flinkSource
             .flatMap(new FlatMapFunction<T, WindowedValue<T>>() {
               @Override
               public void flatMap(T s, Collector<WindowedValue<T>> collector)
throws Exception {

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/aead96ff/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/types/FlinkCoder.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/types/FlinkCoder.java
b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/types/FlinkCoder.java
new file mode 100644
index 0000000..3b1e66e
--- /dev/null
+++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/types/FlinkCoder.java
@@ -0,0 +1,64 @@
+/*
+ * 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.flink.translation.types;
+
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.StandardCoder;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * A Coder that uses Flink's serialization system.
+ * @param <T> The type of the value to be encoded
+ */
+public class FlinkCoder<T> extends StandardCoder<T> {
+
+  private final TypeSerializer<T> typeSerializer;
+
+  public FlinkCoder(TypeInformation<T> typeInformation, ExecutionConfig executionConfig)
{
+    this.typeSerializer = typeInformation.createSerializer(executionConfig);
+  }
+
+  @Override
+  public void encode(T value, OutputStream outStream, Context context) throws IOException
{
+    typeSerializer.serialize(value, new DataOutputViewStreamWrapper(outStream));
+  }
+
+  @Override
+  public T decode(InputStream inStream, Context context) throws IOException {
+    return typeSerializer.deserialize(new DataInputViewStreamWrapper(inStream));
+  }
+
+  @Override
+  public List<? extends Coder<?>> getCoderArguments() {
+    return Collections.emptyList();
+  }
+
+  @Override
+  public void verifyDeterministic() throws NonDeterministicException {
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/aead96ff/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedFlinkSource.java
----------------------------------------------------------------------
diff --git a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedFlinkSource.java
b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedFlinkSource.java
index 05a8c7a..a157b46 100644
--- a/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedFlinkSource.java
+++ b/runners/flink/runner/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedFlinkSource.java
@@ -17,6 +17,7 @@
  */
 package org.apache.beam.runners.flink.translation.wrappers.streaming.io;
 
+import org.apache.beam.runners.flink.translation.types.CoderTypeInformation;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.io.UnboundedSource;
 import org.apache.beam.sdk.options.PipelineOptions;
@@ -37,6 +38,9 @@ public class UnboundedFlinkSource<T> extends UnboundedSource<T,
UnboundedSource.
 
   private final SourceFunction<T> flinkSource;
 
+  /** Coder set during translation */
+  private Coder<T> coder;
+
   public UnboundedFlinkSource(SourceFunction<T> source) {
     flinkSource = Preconditions.checkNotNull(source);
   }
@@ -68,8 +72,12 @@ public class UnboundedFlinkSource<T> extends UnboundedSource<T,
UnboundedSource.
 
   @Override
   public Coder<T> getDefaultOutputCoder() {
-    // The coder is specified in the Flink source
-    return null;
+    // The coder derived from the Flink source
+    return coder;
+  }
+
+  public void setCoder(Coder<T> coder) {
+    this.coder = coder;
   }
 
   /**


Mime
View raw message