You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by mx...@apache.org on 2016/04/29 17:58:58 UTC
[2/4] incubator-beam git commit: fix Flink source coder handling
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 <mx...@apache.org>
Authored: Thu Apr 28 12:00:18 2016 +0200
Committer: Maximilian Michels <mx...@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;
}
/**