You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2019/06/26 03:33:25 UTC
[flink] 05/20: [FLINK-12729] [state-processing-api] Build operator
state index within OnDiskSavepointMetadata
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 88d2e3cc5c31a7703d1deb90f79cd195f67c9686
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Tue Jun 25 18:55:43 2019 +0800
[FLINK-12729] [state-processing-api] Build operator state index within OnDiskSavepointMetadata
This closes #8615.
---
.../apache/flink/state/api/ExistingSavepoint.java | 32 ++++------------------
.../runtime/metadata/OnDiskSavepointMetadata.java | 23 ++++++++++++----
.../api/runtime/metadata/SavepointMetadata.java | 7 +++--
3 files changed, 29 insertions(+), 33 deletions(-)
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/ExistingSavepoint.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/ExistingSavepoint.java
index 255a7e8..15e9d07 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/ExistingSavepoint.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/ExistingSavepoint.java
@@ -27,12 +27,10 @@ import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TupleTypeInfo;
import org.apache.flink.runtime.checkpoint.OperatorState;
-import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.state.api.input.BroadcastStateInputFormat;
import org.apache.flink.state.api.input.ListStateInputFormat;
import org.apache.flink.state.api.input.UnionStateInputFormat;
-import org.apache.flink.state.api.runtime.OperatorIDGenerator;
import org.apache.flink.state.api.runtime.metadata.OnDiskSavepointMetadata;
import org.apache.flink.state.api.runtime.metadata.SavepointMetadata;
import org.apache.flink.util.Preconditions;
@@ -71,7 +69,7 @@ public class ExistingSavepoint {
* @throws IOException If the savepoint path is invalid or the uid does not exist.
*/
public <T> DataSet<T> readListState(String uid, String name, TypeInformation<T> typeInfo) throws IOException {
- OperatorState operatorState = getOperatorState(uid);
+ OperatorState operatorState = metadata.getOperatorState(uid);
ListStateDescriptor<T> descriptor = new ListStateDescriptor<>(name, typeInfo);
ListStateInputFormat<T> inputFormat = new ListStateInputFormat<>(operatorState, descriptor);
return env.createInput(inputFormat, typeInfo);
@@ -95,7 +93,7 @@ public class ExistingSavepoint {
TypeInformation<T> typeInfo,
TypeSerializer<T> serializer) throws IOException {
- OperatorState operatorState = getOperatorState(uid);
+ OperatorState operatorState = metadata.getOperatorState(uid);
ListStateDescriptor<T> descriptor = new ListStateDescriptor<>(name, serializer);
ListStateInputFormat<T> inputFormat = new ListStateInputFormat<>(operatorState, descriptor);
return env.createInput(inputFormat, typeInfo);
@@ -111,7 +109,7 @@ public class ExistingSavepoint {
* @throws IOException If the savepoint path is invalid or the uid does not exist.
*/
public <T> DataSet<T> readUnionState(String uid, String name, TypeInformation<T> typeInfo) throws IOException {
- OperatorState operatorState = getOperatorState(uid);
+ OperatorState operatorState = metadata.getOperatorState(uid);
ListStateDescriptor<T> descriptor = new ListStateDescriptor<>(name, typeInfo);
UnionStateInputFormat<T> inputFormat = new UnionStateInputFormat<>(operatorState, descriptor);
return env.createInput(inputFormat, typeInfo);
@@ -135,7 +133,7 @@ public class ExistingSavepoint {
TypeInformation<T> typeInfo,
TypeSerializer<T> serializer) throws IOException {
- OperatorState operatorState = getOperatorState(uid);
+ OperatorState operatorState = metadata.getOperatorState(uid);
ListStateDescriptor<T> descriptor = new ListStateDescriptor<>(name, serializer);
UnionStateInputFormat<T> inputFormat = new UnionStateInputFormat<>(operatorState, descriptor);
return env.createInput(inputFormat, typeInfo);
@@ -158,7 +156,7 @@ public class ExistingSavepoint {
TypeInformation<K> keyTypeInfo,
TypeInformation<V> valueTypeInfo) throws IOException {
- OperatorState operatorState = getOperatorState(uid);
+ OperatorState operatorState = metadata.getOperatorState(uid);
MapStateDescriptor<K, V> descriptor = new MapStateDescriptor<>(name, keyTypeInfo, valueTypeInfo);
BroadcastStateInputFormat<K, V> inputFormat = new BroadcastStateInputFormat<>(operatorState, descriptor);
return env.createInput(inputFormat, new TupleTypeInfo<>(keyTypeInfo, valueTypeInfo));
@@ -187,27 +185,9 @@ public class ExistingSavepoint {
TypeSerializer<K> keySerializer,
TypeSerializer<V> valueSerializer) throws IOException {
- OperatorState operatorState = getOperatorState(uid);
+ OperatorState operatorState = metadata.getOperatorState(uid);
MapStateDescriptor<K, V> descriptor = new MapStateDescriptor<>(name, keySerializer, valueSerializer);
BroadcastStateInputFormat<K, V> inputFormat = new BroadcastStateInputFormat<>(operatorState, descriptor);
return env.createInput(inputFormat, new TupleTypeInfo<>(keyTypeInfo, valueTypeInfo));
}
-
- /**
- * Finds the {@link OperatorState} for a uid within a savepoint.
- *
- * @return A handle to the operator state in the savepoint with the provided uid.
- * @throws IOException If the savepoint path is invalid or the uid does not exist.
- */
- private OperatorState getOperatorState(String uid) throws IOException {
- OperatorID operatorID = OperatorIDGenerator.fromUid(uid);
-
- for (final OperatorState state : metadata.getOperatorStates()) {
- if (state.getOperatorID().equals(operatorID)) {
- return state;
- }
- }
-
- throw new IOException("Savepoint does not contain state with operator uid " + uid);
- }
}
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/metadata/OnDiskSavepointMetadata.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/metadata/OnDiskSavepointMetadata.java
index b798f5a..c4010d6 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/metadata/OnDiskSavepointMetadata.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/metadata/OnDiskSavepointMetadata.java
@@ -22,11 +22,15 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.runtime.checkpoint.MasterState;
import org.apache.flink.runtime.checkpoint.OperatorState;
import org.apache.flink.runtime.checkpoint.savepoint.Savepoint;
+import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.state.api.runtime.OperatorIDGenerator;
import org.apache.flink.state.api.runtime.SavepointLoader;
import java.io.IOException;
import java.util.Collection;
import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
/**
* Returns metadata about an existing savepoint.
@@ -40,7 +44,7 @@ public class OnDiskSavepointMetadata implements SavepointMetadata {
private final Collection<MasterState> masterStates;
- private final Collection<OperatorState> operatorStates;
+ private final Map<OperatorID, OperatorState> operatorStateIndex;
public OnDiskSavepointMetadata(String path) throws IOException {
Savepoint savepoint = SavepointLoader.loadSavepoint(path);
@@ -54,7 +58,9 @@ public class OnDiskSavepointMetadata implements SavepointMetadata {
this.masterStates = savepoint.getMasterStates();
- this.operatorStates = savepoint.getOperatorStates();
+ Collection<OperatorState> operatorStates = savepoint.getOperatorStates();
+ this.operatorStateIndex = new HashMap<>(operatorStates.size());
+ operatorStates.forEach(state -> this.operatorStateIndex.put(state.getOperatorID(), state));
}
/**
@@ -74,10 +80,17 @@ public class OnDiskSavepointMetadata implements SavepointMetadata {
}
/**
- * @return Operator states for the savepoint.
+ * @return Operator state for the given UID.
*/
@Override
- public Collection<OperatorState> getOperatorStates() {
- return operatorStates;
+ public OperatorState getOperatorState(String uid) throws IOException {
+ OperatorID operatorID = OperatorIDGenerator.fromUid(uid);
+
+ OperatorState operatorState = operatorStateIndex.get(operatorID);
+ if (operatorState == null) {
+ throw new IOException("Savepoint does not contain state with operator uid " + uid);
+ }
+
+ return operatorState;
}
}
diff --git a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/metadata/SavepointMetadata.java b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/metadata/SavepointMetadata.java
index 29a96b9..3a2b7cb 100644
--- a/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/metadata/SavepointMetadata.java
+++ b/flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/metadata/SavepointMetadata.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.runtime.checkpoint.MasterState;
import org.apache.flink.runtime.checkpoint.OperatorState;
+import java.io.IOException;
import java.io.Serializable;
import java.util.Collection;
@@ -42,7 +43,9 @@ public interface SavepointMetadata extends Serializable {
Collection<MasterState> getMasterStates();
/**
- * @return Operator states for the savepoint.
+ * @return Operator state for the given UID.
+ *
+ * @throws IOException If the savepoint does not contain operator state with the given uid.
*/
- Collection<OperatorState> getOperatorStates();
+ OperatorState getOperatorState(String uid) throws IOException;
}