beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aljos...@apache.org
Subject [1/2] beam git commit: [BEAM-2401] Update Flink Runner to Flink 1.3.0
Date Sun, 04 Jun 2017 06:31:08 GMT
Repository: beam
Updated Branches:
  refs/heads/master ef56ea495 -> 1866a0113


[BEAM-2401] Update Flink Runner to Flink 1.3.0


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

Branch: refs/heads/master
Commit: fbc6cc59bff93dfcf8676f874870a43eeb228c15
Parents: ef56ea4
Author: JingsongLi <lzljs3620320@aliyun.com>
Authored: Fri Jun 2 11:31:44 2017 +0800
Committer: Aljoscha Krettek <aljoscha.krettek@gmail.com>
Committed: Sun Jun 4 08:18:20 2017 +0200

----------------------------------------------------------------------
 runners/flink/pom.xml                           |  2 +-
 .../translation/types/CoderTypeSerializer.java  | 55 ++++++++++++++++++++
 .../types/EncodedValueSerializer.java           | 18 +------
 .../state/FlinkBroadcastStateInternals.java     | 29 +++++------
 .../streaming/UnboundedSourceWrapperTest.java   | 37 +++++++++++++
 5 files changed, 109 insertions(+), 32 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/fbc6cc59/runners/flink/pom.xml
----------------------------------------------------------------------
diff --git a/runners/flink/pom.xml b/runners/flink/pom.xml
index fb0a67c..92f95a0 100644
--- a/runners/flink/pom.xml
+++ b/runners/flink/pom.xml
@@ -31,7 +31,7 @@
   <packaging>jar</packaging>
 
   <properties>
-    <flink.version>1.2.1</flink.version>
+    <flink.version>1.3.0</flink.version>
   </properties>
 
   <profiles>

http://git-wip-us.apache.org/repos/asf/beam/blob/fbc6cc59/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializer.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializer.java
b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializer.java
index e003119..bea562e 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializer.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/types/CoderTypeSerializer.java
@@ -24,7 +24,9 @@ import org.apache.beam.runners.flink.translation.wrappers.DataOutputViewWrapper;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.beam.sdk.coders.CoderException;
 import org.apache.beam.sdk.util.CoderUtils;
+import org.apache.flink.api.common.typeutils.CompatibilityResult;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.TypeSerializerConfigSnapshot;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
 
@@ -129,4 +131,57 @@ public class CoderTypeSerializer<T> extends TypeSerializer<T>
{
   public int hashCode() {
     return coder.hashCode();
   }
+
+  @Override
+  public TypeSerializerConfigSnapshot snapshotConfiguration() {
+    return new CoderTypeSerializerConfigSnapshot<>(coder);
+  }
+
+  @Override
+  public CompatibilityResult<T> ensureCompatibility(TypeSerializerConfigSnapshot configSnapshot)
{
+    if (configSnapshot instanceof CoderTypeSerializerConfigSnapshot) {
+      if (coder.equals(((CoderTypeSerializerConfigSnapshot<?>) configSnapshot).coder))
{
+        return CompatibilityResult.compatible();
+      }
+    }
+    return CompatibilityResult.requiresMigration();
+  }
+
+  /**
+   *  TypeSerializerConfigSnapshot of CoderTypeSerializer.
+   */
+  public static class CoderTypeSerializerConfigSnapshot<T> extends TypeSerializerConfigSnapshot
{
+
+    private static final int VERSION = 1;
+    private Coder<T> coder;
+
+    public CoderTypeSerializerConfigSnapshot(Coder<T> coder) {
+      this.coder = coder;
+    }
+
+    @Override
+    public int getVersion() {
+      return VERSION;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) {
+        return true;
+      }
+      if (o == null || getClass() != o.getClass()) {
+        return false;
+      }
+
+      CoderTypeSerializerConfigSnapshot<?> that = (CoderTypeSerializerConfigSnapshot<?>)
o;
+
+      return coder != null ? coder.equals(that.coder) : that.coder == null;
+    }
+
+    @Override
+    public int hashCode() {
+      return coder.hashCode();
+    }
+  }
+
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/fbc6cc59/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/types/EncodedValueSerializer.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/types/EncodedValueSerializer.java
b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/types/EncodedValueSerializer.java
index c3b9794..c40eb46 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/types/EncodedValueSerializer.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/types/EncodedValueSerializer.java
@@ -20,13 +20,14 @@ package org.apache.beam.runners.flink.translation.types;
 import java.io.IOException;
 import org.apache.beam.sdk.coders.Coder;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.TypeSerializerSingleton;
 import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
 
 /**
  * {@link TypeSerializer} for values that were encoded using a {@link Coder}.
  */
-public final class EncodedValueSerializer extends TypeSerializer<byte[]> {
+public final class EncodedValueSerializer extends TypeSerializerSingleton<byte[]> {
 
   private static final long serialVersionUID = 1L;
 
@@ -57,7 +58,6 @@ public final class EncodedValueSerializer extends TypeSerializer<byte[]>
{
     return -1;
   }
 
-
   @Override
   public void serialize(byte[] record, DataOutputView target) throws IOException {
     if (record == null) {
@@ -94,18 +94,4 @@ public final class EncodedValueSerializer extends TypeSerializer<byte[]>
{
     return obj instanceof EncodedValueSerializer;
   }
 
-  @Override
-  public int hashCode() {
-    return this.getClass().hashCode();
-  }
-
-  @Override
-  public boolean equals(Object obj) {
-    return obj instanceof EncodedValueSerializer;
-  }
-
-  @Override
-  public TypeSerializer<byte[]> duplicate() {
-    return this;
-  }
 }

http://git-wip-us.apache.org/repos/asf/beam/blob/fbc6cc59/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkBroadcastStateInternals.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkBroadcastStateInternals.java
b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkBroadcastStateInternals.java
index f44e668..6cc2429 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkBroadcastStateInternals.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkBroadcastStateInternals.java
@@ -49,11 +49,11 @@ import org.apache.beam.sdk.util.CombineContextFactory;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.state.ListState;
 import org.apache.flink.api.common.state.ListStateDescriptor;
-import org.apache.flink.runtime.state.DefaultOperatorStateBackend;
+import org.apache.flink.api.common.state.OperatorStateStore;
 import org.apache.flink.runtime.state.OperatorStateBackend;
 
 /**
- * {@link StateInternals} that uses a Flink {@link DefaultOperatorStateBackend}
+ * {@link StateInternals} that uses a Flink {@link OperatorStateBackend}
  * to manage the broadcast state.
  * The state is the same on all parallel instances of the operator.
  * So we just need store state of operator-0 in OperatorStateBackend.
@@ -64,13 +64,12 @@ import org.apache.flink.runtime.state.OperatorStateBackend;
 public class FlinkBroadcastStateInternals<K> implements StateInternals {
 
   private int indexInSubtaskGroup;
-  private final DefaultOperatorStateBackend stateBackend;
+  private final OperatorStateBackend stateBackend;
   // stateName -> <namespace, state>
   private Map<String, Map<String, ?>> stateForNonZeroOperator;
 
   public FlinkBroadcastStateInternals(int indexInSubtaskGroup, OperatorStateBackend stateBackend)
{
-    //TODO flink do not yet expose through public API
-    this.stateBackend = (DefaultOperatorStateBackend) stateBackend;
+    this.stateBackend = stateBackend;
     this.indexInSubtaskGroup = indexInSubtaskGroup;
     if (indexInSubtaskGroup != 0) {
       stateForNonZeroOperator = new HashMap<>();
@@ -178,10 +177,10 @@ public class FlinkBroadcastStateInternals<K> implements StateInternals
{
     private String name;
     private final StateNamespace namespace;
     private final ListStateDescriptor<Map<String, T>> flinkStateDescriptor;
-    private final DefaultOperatorStateBackend flinkStateBackend;
+    private final OperatorStateStore flinkStateBackend;
 
     AbstractBroadcastState(
-        DefaultOperatorStateBackend flinkStateBackend,
+        OperatorStateBackend flinkStateBackend,
         String name,
         StateNamespace namespace,
         Coder<T> coder) {
@@ -211,7 +210,7 @@ public class FlinkBroadcastStateInternals<K> implements StateInternals
{
           if (result != null) {
             stateForNonZeroOperator.put(name, result);
             // we don't need it anymore, must clear it.
-            flinkStateBackend.getBroadcastOperatorState(
+            flinkStateBackend.getUnionListState(
                 flinkStateDescriptor).clear();
           }
         }
@@ -220,7 +219,7 @@ public class FlinkBroadcastStateInternals<K> implements StateInternals
{
     }
 
     Map<String, T> getMapFromBroadcastState() throws Exception {
-      ListState<Map<String, T>> state = flinkStateBackend.getBroadcastOperatorState(
+      ListState<Map<String, T>> state = flinkStateBackend.getUnionListState(
           flinkStateDescriptor);
       Iterable<Map<String, T>> iterable = state.get();
       Map<String, T> ret = null;
@@ -239,7 +238,7 @@ public class FlinkBroadcastStateInternals<K> implements StateInternals
{
      */
     void updateMap(Map<String, T> map) throws Exception {
       if (indexInSubtaskGroup == 0) {
-        ListState<Map<String, T>> state = flinkStateBackend.getBroadcastOperatorState(
+        ListState<Map<String, T>> state = flinkStateBackend.getUnionListState(
             flinkStateDescriptor);
         state.clear();
         if (map.size() > 0) {
@@ -304,7 +303,7 @@ public class FlinkBroadcastStateInternals<K> implements StateInternals
{
     private final StateTag<ValueState<T>> address;
 
     FlinkBroadcastValueState(
-        DefaultOperatorStateBackend flinkStateBackend,
+        OperatorStateBackend flinkStateBackend,
         StateTag<ValueState<T>> address,
         StateNamespace namespace,
         Coder<T> coder) {
@@ -365,7 +364,7 @@ public class FlinkBroadcastStateInternals<K> implements StateInternals
{
     private final StateTag<BagState<T>> address;
 
     FlinkBroadcastBagState(
-        DefaultOperatorStateBackend flinkStateBackend,
+        OperatorStateBackend flinkStateBackend,
         StateTag<BagState<T>> address,
         StateNamespace namespace,
         Coder<T> coder) {
@@ -454,7 +453,7 @@ public class FlinkBroadcastStateInternals<K> implements StateInternals
{
     private final Combine.CombineFn<InputT, AccumT, OutputT> combineFn;
 
     FlinkCombiningState(
-        DefaultOperatorStateBackend flinkStateBackend,
+        OperatorStateBackend flinkStateBackend,
         StateTag<CombiningState<InputT, AccumT, OutputT>> address,
         Combine.CombineFn<InputT, AccumT, OutputT> combineFn,
         StateNamespace namespace,
@@ -572,7 +571,7 @@ public class FlinkBroadcastStateInternals<K> implements StateInternals
{
     private final FlinkBroadcastStateInternals<K> flinkStateInternals;
 
     FlinkKeyedCombiningState(
-        DefaultOperatorStateBackend flinkStateBackend,
+        OperatorStateBackend flinkStateBackend,
         StateTag<CombiningState<InputT, AccumT, OutputT>> address,
         Combine.CombineFn<InputT, AccumT, OutputT> combineFn,
         StateNamespace namespace,
@@ -709,7 +708,7 @@ public class FlinkBroadcastStateInternals<K> implements StateInternals
{
     private final CombineWithContext.Context context;
 
     FlinkCombiningStateWithContext(
-        DefaultOperatorStateBackend flinkStateBackend,
+        OperatorStateBackend flinkStateBackend,
         StateTag<CombiningState<InputT, AccumT, OutputT>> address,
         CombineWithContext.CombineFnWithContext<InputT, AccumT, OutputT> combineFn,
         StateNamespace namespace,

http://git-wip-us.apache.org/repos/asf/beam/blob/fbc6cc59/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/UnboundedSourceWrapperTest.java
----------------------------------------------------------------------
diff --git a/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/UnboundedSourceWrapperTest.java
b/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/UnboundedSourceWrapperTest.java
index 500fa66..e3875bc 100644
--- a/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/UnboundedSourceWrapperTest.java
+++ b/runners/flink/src/test/java/org/apache/beam/runners/flink/streaming/UnboundedSourceWrapperTest.java
@@ -55,9 +55,12 @@ import org.apache.flink.streaming.api.operators.StreamSource;
 import org.apache.flink.streaming.api.watermark.Watermark;
 import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
 import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.streamstatus.StreamStatus;
+import org.apache.flink.streaming.runtime.streamstatus.StreamStatusMaintainer;
 import org.apache.flink.streaming.runtime.tasks.StreamTask;
 import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
 import org.apache.flink.util.InstantiationUtil;
+import org.apache.flink.util.OutputTag;
 import org.junit.Test;
 import org.junit.experimental.runners.Enclosed;
 import org.junit.runner.RunWith;
@@ -127,6 +130,7 @@ public class UnboundedSourceWrapperTest {
       try {
         sourceOperator.open();
         sourceOperator.run(checkpointLock,
+            new TestStreamStatusMaintainer(),
             new Output<StreamRecord<WindowedValue<ValueWithRecordId<KV<Integer,
Integer>>>>>() {
               private int count = 0;
 
@@ -135,6 +139,11 @@ public class UnboundedSourceWrapperTest {
               }
 
               @Override
+              public <X> void collect(OutputTag<X> outputTag, StreamRecord<X>
streamRecord) {
+                collect((StreamRecord) streamRecord);
+              }
+
+              @Override
               public void emitLatencyMarker(LatencyMarker latencyMarker) {
               }
 
@@ -215,6 +224,7 @@ public class UnboundedSourceWrapperTest {
       try {
         sourceOperator.open();
         sourceOperator.run(checkpointLock,
+            new TestStreamStatusMaintainer(),
             new Output<StreamRecord<WindowedValue<ValueWithRecordId<KV<Integer,
Integer>>>>>() {
               private int count = 0;
 
@@ -223,6 +233,11 @@ public class UnboundedSourceWrapperTest {
               }
 
               @Override
+              public <X> void collect(OutputTag<X> outputTag, StreamRecord<X>
streamRecord) {
+                collect((StreamRecord) streamRecord);
+              }
+
+              @Override
               public void emitLatencyMarker(LatencyMarker latencyMarker) {
               }
 
@@ -293,6 +308,7 @@ public class UnboundedSourceWrapperTest {
       try {
         restoredSourceOperator.open();
         restoredSourceOperator.run(checkpointLock,
+            new TestStreamStatusMaintainer(),
             new Output<StreamRecord<WindowedValue<ValueWithRecordId<KV<Integer,
Integer>>>>>() {
               private int count = 0;
 
@@ -301,6 +317,11 @@ public class UnboundedSourceWrapperTest {
               }
 
               @Override
+              public <X> void collect(OutputTag<X> outputTag, StreamRecord<X>
streamRecord) {
+                collect((StreamRecord) streamRecord);
+              }
+
+              @Override
               public void emitLatencyMarker(LatencyMarker latencyMarker) {
               }
 
@@ -462,4 +483,20 @@ public class UnboundedSourceWrapperTest {
 
   }
 
+  private static final class TestStreamStatusMaintainer implements StreamStatusMaintainer
{
+    StreamStatus currentStreamStatus = StreamStatus.ACTIVE;
+
+    @Override
+    public void toggleStreamStatus(StreamStatus streamStatus) {
+      if (!currentStreamStatus.equals(streamStatus)) {
+        currentStreamStatus = streamStatus;
+      }
+    }
+
+    @Override
+    public StreamStatus getStreamStatus() {
+      return currentStreamStatus;
+    }
+  }
+
 }


Mime
View raw message