聊聊flink的MemoryStateBackend
序
本文主要研究一下flink的MemoryStateBackendhtml
StateBackend
flink-runtime_2.11-1.7.0-sources.jar!/org/apache/flink/runtime/state/StateBackend.javajava
@PublicEvolving
public interface StateBackend extends java.io.Serializable {
// ------------------------------------------------------------------------
// Checkpoint storage - the durable persistence of checkpoint data
// ------------------------------------------------------------------------
/**
* Resolves the given pointer to a checkpoint/savepoint into a checkpoint location. The location
* supports reading the checkpoint metadata, or disposing the checkpoint storage location.
*
* <p>If the state backend cannot understand the format of the pointer (for example because it
* was created by a different state backend) this method should throw an {@code IOException}.
*
* @param externalPointer The external checkpoint pointer to resolve.
* @return The checkpoint location handle.
*
* @throws IOException Thrown, if the state backend does not understand the pointer, or if
* the pointer could not be resolved due to an I/O error.
*/
CompletedCheckpointStorageLocation resolveCheckpoint(String externalPointer) throws IOException;
/**
* Creates a storage for checkpoints for the given job. The checkpoint storage is
* used to write checkpoint data and metadata.
*
* @param jobId The job to store checkpoint data for.
* @return A checkpoint storage for the given job.
*
* @throws IOException Thrown if the checkpoint storage cannot be initialized.
*/
CheckpointStorage createCheckpointStorage(JobID jobId) throws IOException;
// ------------------------------------------------------------------------
// Structure Backends
// ------------------------------------------------------------------------
/**
* Creates a new {@link AbstractKeyedStateBackend} that is responsible for holding <b>keyed state</b>
* and checkpointing it. Uses default TTL time provider.
*
* <p><i>Keyed State</i> is state where each value is bound to a key.
*
* @param <K> The type of the keys by which the state is organized.
*
* @return The Keyed State Backend for the given job, operator, and key group range.
*
* @throws Exception This method may forward all exceptions that occur while instantiating the backend.
*/
default <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(
Environment env,
JobID jobID,
String operatorIdentifier,
TypeSerializer<K> keySerializer,
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry) throws Exception {
return createKeyedStateBackend(
env,
jobID,
operatorIdentifier,
keySerializer,
numberOfKeyGroups,
keyGroupRange,
kvStateRegistry,
TtlTimeProvider.DEFAULT
);
}
/**
* Creates a new {@link AbstractKeyedStateBackend} that is responsible for holding <b>keyed state</b>
* and checkpointing it.
*
* <p><i>Keyed State</i> is state where each value is bound to a key.
*
* @param <K> The type of the keys by which the state is organized.
*
* @return The Keyed State Backend for the given job, operator, and key group range.
*
* @throws Exception This method may forward all exceptions that occur while instantiating the backend.
*/
default <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(
Environment env,
JobID jobID,
String operatorIdentifier,
TypeSerializer<K> keySerializer,
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry,
TtlTimeProvider ttlTimeProvider
) throws Exception {
return createKeyedStateBackend(
env,
jobID,
operatorIdentifier,
keySerializer,
numberOfKeyGroups,
keyGroupRange,
kvStateRegistry,
ttlTimeProvider,
new UnregisteredMetricsGroup());
}
/**
* Creates a new {@link AbstractKeyedStateBackend} that is responsible for holding <b>keyed state</b>
* and checkpointing it.
*
* <p><i>Keyed State</i> is state where each value is bound to a key.
*
* @param <K> The type of the keys by which the state is organized.
*
* @return The Keyed State Backend for the given job, operator, and key group range.
*
* @throws Exception This method may forward all exceptions that occur while instantiating the backend.
*/
<K> AbstractKeyedStateBackend<K> createKeyedStateBackend(
Environment env,
JobID jobID,
String operatorIdentifier,
TypeSerializer<K> keySerializer,
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry,
TtlTimeProvider ttlTimeProvider,
MetricGroup metricGroup) throws Exception;
/**
* Creates a new {@link OperatorStateBackend} that can be used for storing operator state.
*
* <p>Operator state is state that is associated with parallel operator (or function) instances,
* rather than with keys.
*
* @param env The runtime environment of the executing task.
* @param operatorIdentifier The identifier of the operator whose state should be stored.
*
* @return The OperatorStateBackend for operator identified by the job and operator identifier.
*
* @throws Exception This method may forward all exceptions that occur while instantiating the backend.
*/
OperatorStateBackend createOperatorStateBackend(Environment env, String operatorIdentifier) throws Exception;
}
- StateBackend接口定義了有狀態的streaming應用的state是如何stored以及checkpointed
- flink目前內置支持MemoryStateBackend、FsStateBackend、RocksDBStateBackend三種，若是沒有配置默認爲MemoryStateBackend；在flink-conf.yaml裏頭能夠進行全局的默認配置，不過具體每一個job還能夠經過StreamExecutionEnvironment.setStateBackend來覆蓋全局的配置
- MemoryStateBackend能夠在構造器中指定大小，默認是5MB，能夠增大可是不能超過akka frame size；FsStateBackend模式把TaskManager的state存儲在內存，可是能夠把checkpoint的state存儲到filesystem中(
好比HDFS)；RocksDBStateBackend把working state存儲在RocksDB中，checkpoint的state存儲在filesystem - StateBackend接口定義了createCheckpointStorage、createKeyedStateBackend、createOperatorStateBackend方法；同時繼承了Serializable接口；StateBackend接口的實現要求是線程安全的
- StateBackend有個直接實現的抽象類AbstractStateBackend，而AbstractFileStateBackend及RocksDBStateBackend繼承了AbstractStateBackend，以後MemoryStateBackend、FsStateBackend都繼承了AbstractFileStateBackend
AbstractStateBackend
flink-runtime_2.11-1.7.0-sources.jar!/org/apache/flink/runtime/state/AbstractStateBackend.javaapache
/**
* An abstract base implementation of the {@link StateBackend} interface.
*
* <p>This class has currently no contents and only kept to not break the prior class hierarchy for users.
*/
@PublicEvolving
public abstract class AbstractStateBackend implements StateBackend, java.io.Serializable {
private static final long serialVersionUID = 4620415814639230247L;
// ------------------------------------------------------------------------
// State Backend - State-Holding Backends
// ------------------------------------------------------------------------
@Override
public abstract <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(
Environment env,
JobID jobID,
String operatorIdentifier,
TypeSerializer<K> keySerializer,
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry,
TtlTimeProvider ttlTimeProvider,
MetricGroup metricGroup) throws IOException;
@Override
public abstract OperatorStateBackend createOperatorStateBackend(
Environment env,
String operatorIdentifier) throws Exception;
}
- AbstractStateBackend聲明實現StateBackend及Serializable接口，這裏沒有新增其餘內容
AbstractFileStateBackend
flink-runtime_2.11-1.7.0-sources.jar!/org/apache/flink/runtime/state/filesystem/AbstractFileStateBackend.java安全
@PublicEvolving
public abstract class AbstractFileStateBackend extends AbstractStateBackend {
private static final long serialVersionUID = 1L;
// ------------------------------------------------------------------------
// State Backend Properties
// ------------------------------------------------------------------------
/** The path where checkpoints will be stored, or null, if none has been configured. */
@Nullable
private final Path baseCheckpointPath;
/** The path where savepoints will be stored, or null, if none has been configured. */
@Nullable
private final Path baseSavepointPath;
//......
// ------------------------------------------------------------------------
// Initialization and metadata storage
// ------------------------------------------------------------------------
@Override
public CompletedCheckpointStorageLocation resolveCheckpoint(String pointer) throws IOException {
return AbstractFsCheckpointStorage.resolveCheckpointPointer(pointer);
}
// ------------------------------------------------------------------------
// Utilities
// ------------------------------------------------------------------------
/**
* Checks the validity of the path's scheme and path.
*
* @param path The path to check.
* @return The URI as a Path.
*
* @throws IllegalArgumentException Thrown, if the URI misses scheme or path.
*/
private static Path validatePath(Path path) {
final URI uri = path.toUri();
final String scheme = uri.getScheme();
final String pathPart = uri.getPath();
// some validity checks
if (scheme == null) {
throw new IllegalArgumentException("The scheme (hdfs://, file://, etc) is null. " +
"Please specify the file system scheme explicitly in the URI.");
}
if (pathPart == null) {
throw new IllegalArgumentException("The path to store the checkpoint data in is null. " +
"Please specify a directory path for the checkpoint data.");
}
if (pathPart.length() == 0 || pathPart.equals("/")) {
throw new IllegalArgumentException("Cannot use the root directory for checkpoints.");
}
return path;
}
@Nullable
private static Path parameterOrConfigured(@Nullable Path path, Configuration config, ConfigOption<String> option) {
if (path != null) {
return path;
}
else {
String configValue = config.getString(option);
try {
return configValue == null ? null : new Path(configValue);
}
catch (IllegalArgumentException e) {
throw new IllegalConfigurationException("Cannot parse value for " + option.key() +
" : " + configValue + " . Not a valid path.");
}
}
}
}
- AbstractFileStateBackend繼承了AbstractStateBackend，它有baseCheckpointPath、baseSavepointPath兩個屬性，容許爲null，路徑的格式爲hdfs://或者file://開頭；resolveCheckpoint方法用於解析checkpoint或savepoint的location，這裏使用AbstractFsCheckpointStorage.resolveCheckpointPointer(pointer)來完成
MemoryStateBackend
flink-runtime_2.11-1.7.0-sources.jar!/org/apache/flink/runtime/state/memory/MemoryStateBackend.javaasync
@PublicEvolving
public class MemoryStateBackend extends AbstractFileStateBackend implements ConfigurableStateBackend {
private static final long serialVersionUID = 4109305377809414635L;
/** The default maximal size that the snapshotted memory state may have (5 MiBytes). */
public static final int DEFAULT_MAX_STATE_SIZE = 5 * 1024 * 1024;
/** The maximal size that the snapshotted memory state may have. */
private final int maxStateSize;
/** Switch to chose between synchronous and asynchronous snapshots.
* A value of 'UNDEFINED' means not yet configured, in which case the default will be used. */
private final TernaryBoolean asynchronousSnapshots;
// ------------------------------------------------------------------------
/**
* Creates a new memory state backend that accepts states whose serialized forms are
* up to the default state size (5 MB).
*
* <p>Checkpoint and default savepoint locations are used as specified in the
* runtime configuration.
*/
public MemoryStateBackend() {
this(null, null, DEFAULT_MAX_STATE_SIZE, TernaryBoolean.UNDEFINED);
}
/**
* Creates a new memory state backend that accepts states whose serialized forms are
* up to the default state size (5 MB). The state backend uses asynchronous snapshots
* or synchronous snapshots as configured.
*
* <p>Checkpoint and default savepoint locations are used as specified in the
* runtime configuration.
*
* @param asynchronousSnapshots Switch to enable asynchronous snapshots.
*/
public MemoryStateBackend(boolean asynchronousSnapshots) {
this(null, null, DEFAULT_MAX_STATE_SIZE, TernaryBoolean.fromBoolean(asynchronousSnapshots));
}
/**
* Creates a new memory state backend that accepts states whose serialized forms are
* up to the given number of bytes.
*
* <p>Checkpoint and default savepoint locations are used as specified in the
* runtime configuration.
*
* <p><b>WARNING:</b> Increasing the size of this value beyond the default value
* ({@value #DEFAULT_MAX_STATE_SIZE}) should be done with care.
* The checkpointed state needs to be send to the JobManager via limited size RPC messages, and there
* and the JobManager needs to be able to hold all aggregated state in its memory.
*
* @param maxStateSize The maximal size of the serialized state
*/
public MemoryStateBackend(int maxStateSize) {
this(null, null, maxStateSize, TernaryBoolean.UNDEFINED);
}
/**
* Creates a new memory state backend that accepts states whose serialized forms are
* up to the given number of bytes and that uses asynchronous snashots as configured.
*
* <p>Checkpoint and default savepoint locations are used as specified in the
* runtime configuration.
*
* <p><b>WARNING:</b> Increasing the size of this value beyond the default value
* ({@value #DEFAULT_MAX_STATE_SIZE}) should be done with care.
* The checkpointed state needs to be send to the JobManager via limited size RPC messages, and there
* and the JobManager needs to be able to hold all aggregated state in its memory.
*
* @param maxStateSize The maximal size of the serialized state
* @param asynchronousSnapshots Switch to enable asynchronous snapshots.
*/
public MemoryStateBackend(int maxStateSize, boolean asynchronousSnapshots) {
this(null, null, maxStateSize, TernaryBoolean.fromBoolean(asynchronousSnapshots));
}
/**
* Creates a new MemoryStateBackend, setting optionally the path to persist checkpoint metadata
* to, and to persist savepoints to.
*
* @param checkpointPath The path to write checkpoint metadata to. If null, the value from
* the runtime configuration will be used.
* @param savepointPath The path to write savepoints to. If null, the value from
* the runtime configuration will be used.
*/
public MemoryStateBackend(@Nullable String checkpointPath, @Nullable String savepointPath) {
this(checkpointPath, savepointPath, DEFAULT_MAX_STATE_SIZE, TernaryBoolean.UNDEFINED);
}
/**
* Creates a new MemoryStateBackend, setting optionally the paths to persist checkpoint metadata
* and savepoints to, as well as configuring state thresholds and asynchronous operations.
*
* <p><b>WARNING:</b> Increasing the size of this value beyond the default value
* ({@value #DEFAULT_MAX_STATE_SIZE}) should be done with care.
* The checkpointed state needs to be send to the JobManager via limited size RPC messages, and there
* and the JobManager needs to be able to hold all aggregated state in its memory.
*
* @param checkpointPath The path to write checkpoint metadata to. If null, the value from
* the runtime configuration will be used.
* @param savepointPath The path to write savepoints to. If null, the value from
* the runtime configuration will be used.
* @param maxStateSize The maximal size of the serialized state.
* @param asynchronousSnapshots Flag to switch between synchronous and asynchronous
* snapshot mode. If null, the value configured in the
* runtime configuration will be used.
*/
public MemoryStateBackend(
@Nullable String checkpointPath,
@Nullable String savepointPath,
int maxStateSize,
TernaryBoolean asynchronousSnapshots) {
super(checkpointPath == null ? null : new Path(checkpointPath),
savepointPath == null ? null : new Path(savepointPath));
checkArgument(maxStateSize > 0, "maxStateSize must be > 0");
this.maxStateSize = maxStateSize;
this.asynchronousSnapshots = asynchronousSnapshots;
}
/**
* Private constructor that creates a re-configured copy of the state backend.
*
* @param original The state backend to re-configure
* @param configuration The configuration
*/
private MemoryStateBackend(MemoryStateBackend original, Configuration configuration) {
super(original.getCheckpointPath(), original.getSavepointPath(), configuration);
this.maxStateSize = original.maxStateSize;
// if asynchronous snapshots were configured, use that setting,
// else check the configuration
this.asynchronousSnapshots = original.asynchronousSnapshots.resolveUndefined(
configuration.getBoolean(CheckpointingOptions.ASYNC_SNAPSHOTS));
}
// ------------------------------------------------------------------------
// Properties
// ------------------------------------------------------------------------
/**
* Gets the maximum size that an individual state can have, as configured in the
* constructor (by default {@value #DEFAULT_MAX_STATE_SIZE}).
*
* @return The maximum size that an individual state can have
*/
public int getMaxStateSize() {
return maxStateSize;
}
/**
* Gets whether the key/value data structures are asynchronously snapshotted.
*
* <p>If not explicitly configured, this is the default value of
* {@link CheckpointingOptions#ASYNC_SNAPSHOTS}.
*/
public boolean isUsingAsynchronousSnapshots() {
return asynchronousSnapshots.getOrDefault(CheckpointingOptions.ASYNC_SNAPSHOTS.defaultValue());
}
// ------------------------------------------------------------------------
// Reconfiguration
// ------------------------------------------------------------------------
/**
* Creates a copy of this state backend that uses the values defined in the configuration
* for fields where that were not specified in this state backend.
*
* @param config the configuration
* @return The re-configured variant of the state backend
*/
@Override
public MemoryStateBackend configure(Configuration config) {
return new MemoryStateBackend(this, config);
}
// ------------------------------------------------------------------------
// checkpoint state persistence
// ------------------------------------------------------------------------
@Override
public CheckpointStorage createCheckpointStorage(JobID jobId) throws IOException {
return new MemoryBackendCheckpointStorage(jobId, getCheckpointPath(), getSavepointPath(), maxStateSize);
}
// ------------------------------------------------------------------------
// state holding structures
// ------------------------------------------------------------------------
@Override
public OperatorStateBackend createOperatorStateBackend(
Environment env,
String operatorIdentifier) throws Exception {
return new DefaultOperatorStateBackend(
env.getUserClassLoader(),
env.getExecutionConfig(),
isUsingAsynchronousSnapshots());
}
@Override
public <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(
Environment env,
JobID jobID,
String operatorIdentifier,
TypeSerializer<K> keySerializer,
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry,
TtlTimeProvider ttlTimeProvider,
MetricGroup metricGroup) {
TaskStateManager taskStateManager = env.getTaskStateManager();
HeapPriorityQueueSetFactory priorityQueueSetFactory =
new HeapPriorityQueueSetFactory(keyGroupRange, numberOfKeyGroups, 128);
return new HeapKeyedStateBackend<>(
kvStateRegistry,
keySerializer,
env.getUserClassLoader(),
numberOfKeyGroups,
keyGroupRange,
isUsingAsynchronousSnapshots(),
env.getExecutionConfig(),
taskStateManager.createLocalRecoveryConfig(),
priorityQueueSetFactory,
ttlTimeProvider);
}
// ------------------------------------------------------------------------
// utilities
// ------------------------------------------------------------------------
@Override
public String toString() {
return "MemoryStateBackend (data in heap memory / checkpoints to JobManager) " +
"(checkpoints: '" + getCheckpointPath() +
"', savepoints: '" + getSavepointPath() +
"', asynchronous: " + asynchronousSnapshots +
", maxStateSize: " + maxStateSize + ")";
}
}
- MemoryStateBackend繼承了AbstractFileStateBackend，實現ConfigurableStateBackend接口(
configure方法)；它將TaskManager的working state及JobManager的checkpoint state存儲在JVM heap中(可是爲了高可用，也能夠設置checkpoint state存儲到filesystem)；MemoryStateBackend僅僅用來作實驗用途，好比本地啓動或者所需的state很是小，對於生產須要改成使用FsStateBackend(將TaskManager的working state存儲在內存，可是將JobManager的checkpoint state存儲到文件系統以支持更大的state存儲) - MemoryStateBackend有個maxStateSize屬性(
默認DEFAULT_MAX_STATE_SIZE爲5MB)，每一個state的大小不能超過maxStateSize，一個task的全部state不能超過RPC系統的限制(默認是10MB，能夠修改但不建議)，全部retained checkpoints的state大小總和不能超過JobManager的JVM heap大小；另外若是建立MemoryStateBackend時未指定checkpointPath及savepointPath，則會從flink-conf.yaml中讀取全局默認值；MemoryStateBackend裏頭還有一個asynchronousSnapshots屬性，是TernaryBoolean類型(TRUE、FALSE、UNDEFINED)，其中UNDEFINED表示沒有配置，將會使用默認值 - MemoryStateBackend的createCheckpointStorage建立的是MemoryBackendCheckpointStorage；createOperatorStateBackend方法建立的是OperatorStateBackend；createKeyedStateBackend方法建立的是HeapKeyedStateBackend
小結
- StateBackend接口定義了有狀態的streaming應用的state是如何stored以及checkpointed；目前內置支持MemoryStateBackend、FsStateBackend、RocksDBStateBackend三種，若是沒有配置默認爲MemoryStateBackend；在flink-conf.yaml裏頭能夠進行全局的默認配置，不過具體每一個job還能夠經過StreamExecutionEnvironment.setStateBackend來覆蓋全局的配置
- StateBackend接口定義了createCheckpointStorage、createKeyedStateBackend、createOperatorStateBackend方法；同時繼承了Serializable接口；StateBackend接口的實現要求是線程安全的；StateBackend有個直接實現的抽象類AbstractStateBackend，而AbstractFileStateBackend及RocksDBStateBackend繼承了AbstractStateBackend，以後MemoryStateBackend、FsStateBackend都繼承了AbstractFileStateBackend
- MemoryStateBackend繼承了AbstractFileStateBackend，實現ConfigurableStateBackend接口(
configure方法)；它將TaskManager的working state及JobManager的checkpoint state存儲在JVM heap中；MemoryStateBackend的createCheckpointStorage建立的是MemoryBackendCheckpointStorage；createOperatorStateBackend方法建立的是OperatorStateBackend；createKeyedStateBackend方法建立的是HeapKeyedStateBackend
doc
相關文章
- 1. 聊聊flink的MemoryStateBackend
- 2. 聊聊flink的JDBCOutputFormat
- 3. 聊聊flink的BlobWriter
- 4. 聊聊flink的ProcessFunction
- 5. 聊聊flink的AbstractTtlState
- 6. 聊聊flink的PartitionableListState
- 7. 聊聊flink的InputFormatSourceFunction
- 8. 聊聊flink的SourceFunction
- 9. 聊聊flink的ScheduledExecutor
- 10. 聊聊flink的ParallelIteratorInputFormat
- 更多相關文章...
- • PHP imagecolorclosest - 取得與指定的顏色最接近的顏色的索引值 - PHP參考手冊
- • PHP imagecolorclosesthwb - 取得與指定的顏色最接近的色度的黑白色的索引 - PHP參考手冊
- • Flink 數據傳輸及反壓詳解
- • 漫談MySQL的鎖機制
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. 聊聊flink的MemoryStateBackend
- 2. 聊聊flink的JDBCOutputFormat
- 3. 聊聊flink的BlobWriter
- 4. 聊聊flink的ProcessFunction
- 5. 聊聊flink的AbstractTtlState
- 6. 聊聊flink的PartitionableListState
- 7. 聊聊flink的InputFormatSourceFunction
- 8. 聊聊flink的SourceFunction
- 9. 聊聊flink的ScheduledExecutor
- 10. 聊聊flink的ParallelIteratorInputFormat