You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by dw...@apache.org on 2023/03/08 10:28:51 UTC
[flink] branch master updated: [FLINK-30113] Implement state compression for broadcast and regular operator states (#21636)
This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new d04d727958b [FLINK-30113] Implement state compression for broadcast and regular operator states (#21636)
d04d727958b is described below
commit d04d727958b641c041d593618de6a40fd62d5338
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Wed Mar 8 11:28:39 2023 +0100
[FLINK-30113] Implement state compression for broadcast and regular operator states (#21636)
Add a new snapshot format version 6 compatible with the previous versions that adds a boolean for snapshot compression
At snapshot time: depending on user configuration, write Snappy compressed regular and broadcast operator states (but not the snapshot meta info) and write the compression boolean accordingly
At restoration time: depending on the read compression boolean, read the snapshots as compressed or not.
---
.../state/CompressibleFSDataInputStream.java | 61 ++++++++++++++++++++
.../state/CompressibleFSDataOutputStream.java | 66 ++++++++++++++++++++++
.../state/DefaultOperatorStateBackendBuilder.java | 7 ++-
...efaultOperatorStateBackendSnapshotStrategy.java | 55 ++++++++++++------
.../state/OperatorBackendSerializationProxy.java | 24 ++++++--
.../state/OperatorStateRestoreOperation.java | 36 ++++++++----
.../StateMetaInfoSnapshotReadersWriters.java | 4 +-
.../runtime/state/SerializationProxiesTest.java | 3 +-
8 files changed, 221 insertions(+), 35 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataInputStream.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataInputStream.java
new file mode 100644
index 00000000000..b38554eb0e3
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataInputStream.java
@@ -0,0 +1,61 @@
+/*
+ * 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.flink.runtime.state;
+
+import org.apache.flink.core.fs.FSDataInputStream;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * {@link FSDataInputStream} that delegates all reading operations to a wrapping {@link
+ * StreamCompressionDecorator}.
+ */
+public class CompressibleFSDataInputStream extends FSDataInputStream {
+
+ private final FSDataInputStream delegate;
+ private final InputStream compressingDelegate;
+
+ public CompressibleFSDataInputStream(
+ FSDataInputStream delegate, StreamCompressionDecorator compressionDecorator)
+ throws IOException {
+ this.delegate = delegate;
+ this.compressingDelegate = compressionDecorator.decorateWithCompression(delegate);
+ }
+
+ @Override
+ public void seek(long desired) throws IOException {
+ delegate.seek(desired);
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return delegate.getPos();
+ }
+
+ @Override
+ public int read() throws IOException {
+ return compressingDelegate.read();
+ }
+
+ @Override
+ public void close() throws IOException {
+ compressingDelegate.close();
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataOutputStream.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataOutputStream.java
new file mode 100644
index 00000000000..a573417085a
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/CompressibleFSDataOutputStream.java
@@ -0,0 +1,66 @@
+/*
+ * 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.flink.runtime.state;
+
+import org.apache.flink.core.fs.FSDataOutputStream;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+/**
+ * {@link FSDataOutputStream} that delegates all writing operations to a wrapping {@link
+ * StreamCompressionDecorator}.
+ */
+public class CompressibleFSDataOutputStream extends FSDataOutputStream {
+
+ private final FSDataOutputStream delegate;
+ private final OutputStream compressingDelegate;
+
+ public CompressibleFSDataOutputStream(
+ CheckpointStateOutputStream delegate, StreamCompressionDecorator compressionDecorator)
+ throws IOException {
+ this.delegate = delegate;
+ this.compressingDelegate = compressionDecorator.decorateWithCompression(delegate);
+ }
+
+ @Override
+ public long getPos() throws IOException {
+ return delegate.getPos();
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ compressingDelegate.write(b);
+ }
+
+ @Override
+ public void flush() throws IOException {
+ compressingDelegate.flush();
+ }
+
+ @Override
+ public void sync() throws IOException {
+ delegate.sync();
+ }
+
+ @Override
+ public void close() throws IOException {
+ compressingDelegate.close();
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendBuilder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendBuilder.java
index bc72ab31464..315543a4ad5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendBuilder.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendBuilder.java
@@ -66,9 +66,14 @@ public class DefaultOperatorStateBackendBuilder
new HashMap<>();
CloseableRegistry cancelStreamRegistryForBackend = new CloseableRegistry();
+ final StreamCompressionDecorator compressionDecorator =
+ AbstractStateBackend.getCompressionDecorator(executionConfig);
DefaultOperatorStateBackendSnapshotStrategy snapshotStrategy =
new DefaultOperatorStateBackendSnapshotStrategy(
- userClassloader, registeredOperatorStates, registeredBroadcastStates);
+ userClassloader,
+ registeredOperatorStates,
+ registeredBroadcastStates,
+ compressionDecorator);
OperatorStateRestoreOperation restoreOperation =
new OperatorStateRestoreOperation(
cancelStreamRegistry,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendSnapshotStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendSnapshotStrategy.java
index 6dc673999dd..2351f1ea119 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendSnapshotStrategy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackendSnapshotStrategy.java
@@ -31,24 +31,32 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
-/** Snapshot strategy for this backend. */
+/**
+ * Snapshot strategy for this backend. This strategy compresses the regular and broadcast operator
+ * states if enabled by configuration.
+ */
class DefaultOperatorStateBackendSnapshotStrategy
implements SnapshotStrategy<
OperatorStateHandle,
DefaultOperatorStateBackendSnapshotStrategy
.DefaultOperatorStateBackendSnapshotResources> {
+
private final ClassLoader userClassLoader;
private final Map<String, PartitionableListState<?>> registeredOperatorStates;
private final Map<String, BackendWritableBroadcastState<?, ?>> registeredBroadcastStates;
+ private final StreamCompressionDecorator compressionDecorator;
protected DefaultOperatorStateBackendSnapshotStrategy(
ClassLoader userClassLoader,
Map<String, PartitionableListState<?>> registeredOperatorStates,
- Map<String, BackendWritableBroadcastState<?, ?>> registeredBroadcastStates) {
+ Map<String, BackendWritableBroadcastState<?, ?>> registeredBroadcastStates,
+ StreamCompressionDecorator compressionDecorator) {
this.userClassLoader = userClassLoader;
this.registeredOperatorStates = registeredOperatorStates;
this.registeredBroadcastStates = registeredBroadcastStates;
+ this.compressionDecorator = compressionDecorator;
}
@Override
@@ -145,7 +153,11 @@ class DefaultOperatorStateBackendSnapshotStrategy
OperatorBackendSerializationProxy backendSerializationProxy =
new OperatorBackendSerializationProxy(
- operatorMetaInfoSnapshots, broadcastMetaInfoSnapshots);
+ operatorMetaInfoSnapshots,
+ broadcastMetaInfoSnapshots,
+ !Objects.equals(
+ UncompressedStreamCompressionDecorator.INSTANCE,
+ compressionDecorator));
backendSerializationProxy.write(dov);
@@ -162,11 +174,18 @@ class DefaultOperatorStateBackendSnapshotStrategy
registeredOperatorStatesDeepCopies.entrySet()) {
PartitionableListState<?> value = entry.getValue();
- long[] partitionOffsets = value.write(localOut);
- OperatorStateHandle.Mode mode = value.getStateMetaInfo().getAssignmentMode();
- writtenStatesMetaData.put(
- entry.getKey(),
- new OperatorStateHandle.StateMetaInfo(partitionOffsets, mode));
+ // create the compressed stream for each state to have the compression header for
+ // each
+ try (final CompressibleFSDataOutputStream compressedLocalOut =
+ new CompressibleFSDataOutputStream(
+ localOut,
+ compressionDecorator)) { // closes only the outer compression stream
+ long[] partitionOffsets = value.write(compressedLocalOut);
+ OperatorStateHandle.Mode mode = value.getStateMetaInfo().getAssignmentMode();
+ writtenStatesMetaData.put(
+ entry.getKey(),
+ new OperatorStateHandle.StateMetaInfo(partitionOffsets, mode));
+ }
}
// ... and the broadcast states themselves ...
@@ -174,24 +193,28 @@ class DefaultOperatorStateBackendSnapshotStrategy
registeredBroadcastStatesDeepCopies.entrySet()) {
BackendWritableBroadcastState<?, ?> value = entry.getValue();
- long[] partitionOffsets = {value.write(localOut)};
- OperatorStateHandle.Mode mode = value.getStateMetaInfo().getAssignmentMode();
- writtenStatesMetaData.put(
- entry.getKey(),
- new OperatorStateHandle.StateMetaInfo(partitionOffsets, mode));
+ // create the compressed stream for each state to have the compression header for
+ // each
+ try (final CompressibleFSDataOutputStream compressedLocalOut =
+ new CompressibleFSDataOutputStream(
+ localOut,
+ compressionDecorator)) { // closes only the outer compression stream
+ long[] partitionOffsets = {value.write(compressedLocalOut)};
+ OperatorStateHandle.Mode mode = value.getStateMetaInfo().getAssignmentMode();
+ writtenStatesMetaData.put(
+ entry.getKey(),
+ new OperatorStateHandle.StateMetaInfo(partitionOffsets, mode));
+ }
}
// ... and, finally, create the state handle.
OperatorStateHandle retValue = null;
if (snapshotCloseableRegistry.unregisterCloseable(localOut)) {
-
StreamStateHandle stateHandle = localOut.closeAndGetHandle();
-
if (stateHandle != null) {
retValue = new OperatorStreamStateHandle(writtenStatesMetaData, stateHandle);
}
-
return SnapshotResult.of(retValue);
} else {
throw new IOException("Stream was already unregistered.");
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorBackendSerializationProxy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorBackendSerializationProxy.java
index aa8be910b4e..75e4a54cc39 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorBackendSerializationProxy.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorBackendSerializationProxy.java
@@ -40,7 +40,7 @@ import static org.apache.flink.runtime.state.metainfo.StateMetaInfoSnapshotReade
*/
public class OperatorBackendSerializationProxy extends VersionedIOReadableWritable {
- public static final int VERSION = 5;
+ public static final int VERSION = 6;
private static final Map<Integer, Integer> META_INFO_SNAPSHOT_FORMAT_VERSION_MAPPER =
new HashMap<>();
@@ -50,12 +50,15 @@ public class OperatorBackendSerializationProxy extends VersionedIOReadableWritab
META_INFO_SNAPSHOT_FORMAT_VERSION_MAPPER.put(2, 2);
META_INFO_SNAPSHOT_FORMAT_VERSION_MAPPER.put(3, 3);
META_INFO_SNAPSHOT_FORMAT_VERSION_MAPPER.put(4, 5);
- META_INFO_SNAPSHOT_FORMAT_VERSION_MAPPER.put(5, CURRENT_STATE_META_INFO_SNAPSHOT_VERSION);
+ META_INFO_SNAPSHOT_FORMAT_VERSION_MAPPER.put(5, 6);
+ META_INFO_SNAPSHOT_FORMAT_VERSION_MAPPER.put(6, CURRENT_STATE_META_INFO_SNAPSHOT_VERSION);
}
private List<StateMetaInfoSnapshot> operatorStateMetaInfoSnapshots;
private List<StateMetaInfoSnapshot> broadcastStateMetaInfoSnapshots;
private ClassLoader userCodeClassLoader;
+ /** This specifies if we use a compressed format write the operator states */
+ private boolean usingStateCompression;
public OperatorBackendSerializationProxy(ClassLoader userCodeClassLoader) {
this.userCodeClassLoader = Preconditions.checkNotNull(userCodeClassLoader);
@@ -63,7 +66,8 @@ public class OperatorBackendSerializationProxy extends VersionedIOReadableWritab
public OperatorBackendSerializationProxy(
List<StateMetaInfoSnapshot> operatorStateMetaInfoSnapshots,
- List<StateMetaInfoSnapshot> broadcastStateMetaInfoSnapshots) {
+ List<StateMetaInfoSnapshot> broadcastStateMetaInfoSnapshots,
+ boolean compression) {
this.operatorStateMetaInfoSnapshots =
Preconditions.checkNotNull(operatorStateMetaInfoSnapshots);
@@ -72,6 +76,11 @@ public class OperatorBackendSerializationProxy extends VersionedIOReadableWritab
Preconditions.checkArgument(
operatorStateMetaInfoSnapshots.size() <= Short.MAX_VALUE
&& broadcastStateMetaInfoSnapshots.size() <= Short.MAX_VALUE);
+ this.usingStateCompression = compression;
+ }
+
+ boolean isUsingStateCompression() {
+ return usingStateCompression;
}
@Override
@@ -81,12 +90,14 @@ public class OperatorBackendSerializationProxy extends VersionedIOReadableWritab
@Override
public int[] getCompatibleVersions() {
- return new int[] {VERSION, 4, 3, 2, 1};
+ return new int[] {VERSION, 5, 4, 3, 2, 1};
}
@Override
public void write(DataOutputView out) throws IOException {
super.write(out);
+ // write the compression format used to write each operator state
+ out.writeBoolean(usingStateCompression);
writeStateMetaInfoSnapshots(operatorStateMetaInfoSnapshots, out);
writeStateMetaInfoSnapshots(broadcastStateMetaInfoSnapshots, out);
}
@@ -112,6 +123,11 @@ public class OperatorBackendSerializationProxy extends VersionedIOReadableWritab
"Cannot determine corresponding meta info snapshot version for operator backend serialization readVersion="
+ proxyReadVersion);
}
+ if (metaInfoSnapshotVersion >= 7) {
+ usingStateCompression = in.readBoolean();
+ } else {
+ usingStateCompression = false;
+ }
final StateMetaInfoReader stateMetaInfoReader =
StateMetaInfoSnapshotReadersWriters.getReader(metaInfoSnapshotVersion);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateRestoreOperation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateRestoreOperation.java
index ba268c09a5e..f818eb81978 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateRestoreOperation.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/OperatorStateRestoreOperation.java
@@ -176,17 +176,31 @@ public class OperatorStateRestoreOperation implements RestoreOperation<Void> {
PartitionableListState<?> listStateForName =
registeredOperatorStates.get(stateName);
- if (listStateForName == null) {
- BackendWritableBroadcastState<?, ?> broadcastStateForName =
- registeredBroadcastStates.get(stateName);
- Preconditions.checkState(
- broadcastStateForName != null,
- "Found state without " + "corresponding meta info: " + stateName);
- deserializeBroadcastStateValues(
- broadcastStateForName, in, nameToOffsets.getValue());
- } else {
- deserializeOperatorStateValues(
- listStateForName, in, nameToOffsets.getValue());
+ final StreamCompressionDecorator compressionDecorator =
+ backendSerializationProxy.isUsingStateCompression()
+ ? SnappyStreamCompressionDecorator.INSTANCE
+ : UncompressedStreamCompressionDecorator.INSTANCE;
+ // create the compressed stream for each state to have the compression header
+ // for each
+ try (final CompressibleFSDataInputStream compressedIn =
+ new CompressibleFSDataInputStream(
+ in,
+ compressionDecorator)) { // closes only the outer compression
+ // stream
+ if (listStateForName == null) {
+ BackendWritableBroadcastState<?, ?> broadcastStateForName =
+ registeredBroadcastStates.get(stateName);
+ Preconditions.checkState(
+ broadcastStateForName != null,
+ "Found state without "
+ + "corresponding meta info: "
+ + stateName);
+ deserializeBroadcastStateValues(
+ broadcastStateForName, compressedIn, nameToOffsets.getValue());
+ } else {
+ deserializeOperatorStateValues(
+ listStateForName, compressedIn, nameToOffsets.getValue());
+ }
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/StateMetaInfoSnapshotReadersWriters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/StateMetaInfoSnapshotReadersWriters.java
index 892756fbecd..4ff14486dcd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/StateMetaInfoSnapshotReadersWriters.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/metainfo/StateMetaInfoSnapshotReadersWriters.java
@@ -43,7 +43,7 @@ public class StateMetaInfoSnapshotReadersWriters {
* Current version for the serialization format of {@link StateMetaInfoSnapshotReadersWriters}.
* - v6: since Flink 1.7.x
*/
- public static final int CURRENT_STATE_META_INFO_SNAPSHOT_VERSION = 6;
+ public static final int CURRENT_STATE_META_INFO_SNAPSHOT_VERSION = 7;
/** Returns the writer for {@link StateMetaInfoSnapshot}. */
@Nonnull
@@ -65,7 +65,7 @@ public class StateMetaInfoSnapshotReadersWriters {
readVersion <= CURRENT_STATE_META_INFO_SNAPSHOT_VERSION,
"Unsupported read version for state meta info [%s]",
readVersion);
- if (readVersion < CURRENT_STATE_META_INFO_SNAPSHOT_VERSION) {
+ if (readVersion < 6) {
// versions before 5 still had different state meta info formats between keyed /
// operator state
throw new UnsupportedOperationException(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/state/SerializationProxiesTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/state/SerializationProxiesTest.java
index 45e9424d60f..8791766065d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/state/SerializationProxiesTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/state/SerializationProxiesTest.java
@@ -175,7 +175,7 @@ public class SerializationProxiesTest {
OperatorBackendSerializationProxy serializationProxy =
new OperatorBackendSerializationProxy(
- stateMetaInfoSnapshots, broadcastStateMetaInfoSnapshots);
+ stateMetaInfoSnapshots, broadcastStateMetaInfoSnapshots, true);
byte[] serialized;
try (ByteArrayOutputStreamWithPos out = new ByteArrayOutputStreamWithPos()) {
@@ -191,6 +191,7 @@ public class SerializationProxiesTest {
serializationProxy.read(new DataInputViewStreamWrapper(in));
}
+ Assert.assertTrue(serializationProxy.isUsingStateCompression());
assertEqualStateMetaInfoSnapshotsLists(
stateMetaInfoSnapshots, serializationProxy.getOperatorStateMetaInfoSnapshots());
assertEqualStateMetaInfoSnapshotsLists(