You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by al...@apache.org on 2017/06/04 06:31:08 UTC
[1/2] beam git commit: [BEAM-2401] Update Flink Runner to Flink 1.3.0
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 <lz...@aliyun.com>
Authored: Fri Jun 2 11:31:44 2017 +0800
Committer: Aljoscha Krettek <al...@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;
+ }
+ }
+
}
[2/2] beam git commit: This closes #3284
Posted by al...@apache.org.
This closes #3284
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/1866a011
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/1866a011
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/1866a011
Branch: refs/heads/master
Commit: 1866a01138ac9095aa7d8dfb1d67c955678b393e
Parents: ef56ea4 fbc6cc5
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Sun Jun 4 08:30:30 2017 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Sun Jun 4 08:30:30 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(-)
----------------------------------------------------------------------