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 <[email protected]> Authored: Thu Apr 28 12:00:18 2016 +0200 Committer: Maximilian Michels <[email protected]> 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; } /**
