public class RocksDBStateBackend extends AbstractStateBackend implements ConfigurableStateBackend
RocksDB
. This state backend can
store very large state that exceeds memory and spills to disk.
All key/value state (including windows) is stored in the key/value index of RocksDB. For persistence against loss of machines, checkpoints take a snapshot of the RocksDB database, and persist that snapshot in a file system (by default) or another configurable state backend.
The behavior of the RocksDB instances can be parametrized by setting RocksDB Options
using the methods setPredefinedOptions(PredefinedOptions)
and
setOptions(OptionsFactory)
.
Modifier and Type | Class and Description |
---|---|
static class |
RocksDBStateBackend.PriorityQueueStateType
The options to chose for the type of priority queue state.
|
Constructor and Description |
---|
RocksDBStateBackend(AbstractStateBackend checkpointStreamBackend)
Deprecated.
Use
RocksDBStateBackend(StateBackend) instead. |
RocksDBStateBackend(AbstractStateBackend checkpointStreamBackend,
boolean enableIncrementalCheckpointing)
Deprecated.
Use
RocksDBStateBackend(StateBackend, TernaryBoolean) instead. |
RocksDBStateBackend(StateBackend checkpointStreamBackend)
Creates a new
RocksDBStateBackend that uses the given state backend to store its
checkpoint data streams. |
RocksDBStateBackend(StateBackend checkpointStreamBackend,
TernaryBoolean enableIncrementalCheckpointing)
Creates a new
RocksDBStateBackend that uses the given state backend to store its
checkpoint data streams. |
RocksDBStateBackend(String checkpointDataUri)
Creates a new
RocksDBStateBackend that stores its checkpoint data in the
file system and location defined by the given URI. |
RocksDBStateBackend(String checkpointDataUri,
boolean enableIncrementalCheckpointing)
Creates a new
RocksDBStateBackend that stores its checkpoint data in the
file system and location defined by the given URI. |
RocksDBStateBackend(URI checkpointDataUri)
Creates a new
RocksDBStateBackend that stores its checkpoint data in the
file system and location defined by the given URI. |
RocksDBStateBackend(URI checkpointDataUri,
boolean enableIncrementalCheckpointing)
Creates a new
RocksDBStateBackend that stores its checkpoint data in the
file system and location defined by the given URI. |
Modifier and Type | Method and Description |
---|---|
RocksDBStateBackend |
configure(Configuration config)
Creates a copy of this state backend that uses the values defined in the configuration
for fields where that were not yet specified in this state backend.
|
CheckpointStorage |
createCheckpointStorage(JobID jobId)
Creates a storage for checkpoints for the given job.
|
<K> AbstractKeyedStateBackend<K> |
createKeyedStateBackend(Environment env,
JobID jobID,
String operatorIdentifier,
TypeSerializer<K> keySerializer,
int numberOfKeyGroups,
KeyGroupRange keyGroupRange,
TaskKvStateRegistry kvStateRegistry,
TtlTimeProvider ttlTimeProvider)
Creates a new
AbstractKeyedStateBackend that is responsible for holding keyed state
and checkpointing it. |
OperatorStateBackend |
createOperatorStateBackend(Environment env,
String operatorIdentifier)
Creates a new
OperatorStateBackend that can be used for storing operator state. |
StateBackend |
getCheckpointBackend()
Gets the state backend that this RocksDB state backend uses to persist
its bytes to.
|
org.rocksdb.ColumnFamilyOptions |
getColumnOptions()
Gets the RocksDB
ColumnFamilyOptions to be used for all RocksDB instances. |
org.rocksdb.DBOptions |
getDbOptions()
Gets the RocksDB
DBOptions to be used for all RocksDB instances. |
String[] |
getDbStoragePaths()
Gets the configured local DB storage paths, or null, if none were configured.
|
OptionsFactory |
getOptions()
Gets the options factory that lazily creates the RocksDB options.
|
PredefinedOptions |
getPredefinedOptions()
Gets the currently set predefined options for RocksDB.
|
boolean |
isIncrementalCheckpointsEnabled()
Gets whether incremental checkpoints are enabled for this state backend.
|
CompletedCheckpointStorageLocation |
resolveCheckpoint(String pointer)
Resolves the given pointer to a checkpoint/savepoint into a checkpoint location.
|
void |
setDbStoragePath(String path)
Sets the path where the RocksDB local database files should be stored on the local
file system.
|
void |
setDbStoragePaths(String... paths)
Sets the directories in which the local RocksDB database puts its files (like SST and
metadata files).
|
void |
setOptions(OptionsFactory optionsFactory)
Sets
Options for the RocksDB instances. |
void |
setPredefinedOptions(PredefinedOptions options)
Sets the predefined options for RocksDB.
|
String |
toString() |
clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait
createKeyedStateBackend
public RocksDBStateBackend(String checkpointDataUri) throws IOException
RocksDBStateBackend
that stores its checkpoint data in the
file system and location defined by the given URI.
A state backend that stores checkpoints in HDFS or S3 must specify the file system host and port in the URI, or have the Hadoop configuration that describes the file system (host / high-availability group / possibly credentials) either referenced from the Flink config, or included in the classpath.
checkpointDataUri
- The URI describing the filesystem and path to the checkpoint data directory.IOException
- Thrown, if no file system can be found for the scheme in the URI.public RocksDBStateBackend(String checkpointDataUri, boolean enableIncrementalCheckpointing) throws IOException
RocksDBStateBackend
that stores its checkpoint data in the
file system and location defined by the given URI.
A state backend that stores checkpoints in HDFS or S3 must specify the file system host and port in the URI, or have the Hadoop configuration that describes the file system (host / high-availability group / possibly credentials) either referenced from the Flink config, or included in the classpath.
checkpointDataUri
- The URI describing the filesystem and path to the checkpoint data directory.enableIncrementalCheckpointing
- True if incremental checkpointing is enabled.IOException
- Thrown, if no file system can be found for the scheme in the URI.public RocksDBStateBackend(URI checkpointDataUri) throws IOException
RocksDBStateBackend
that stores its checkpoint data in the
file system and location defined by the given URI.
A state backend that stores checkpoints in HDFS or S3 must specify the file system host and port in the URI, or have the Hadoop configuration that describes the file system (host / high-availability group / possibly credentials) either referenced from the Flink config, or included in the classpath.
checkpointDataUri
- The URI describing the filesystem and path to the checkpoint data directory.IOException
- Thrown, if no file system can be found for the scheme in the URI.public RocksDBStateBackend(URI checkpointDataUri, boolean enableIncrementalCheckpointing) throws IOException
RocksDBStateBackend
that stores its checkpoint data in the
file system and location defined by the given URI.
A state backend that stores checkpoints in HDFS or S3 must specify the file system host and port in the URI, or have the Hadoop configuration that describes the file system (host / high-availability group / possibly credentials) either referenced from the Flink config, or included in the classpath.
checkpointDataUri
- The URI describing the filesystem and path to the checkpoint data directory.enableIncrementalCheckpointing
- True if incremental checkpointing is enabled.IOException
- Thrown, if no file system can be found for the scheme in the URI.public RocksDBStateBackend(StateBackend checkpointStreamBackend)
RocksDBStateBackend
that uses the given state backend to store its
checkpoint data streams. Typically, one would supply a filesystem or database state backend
here where the snapshots from RocksDB would be stored.
The snapshots of the RocksDB state will be stored using the given backend's
StateBackend.createCheckpointStorage(JobID)
.
checkpointStreamBackend
- The backend write the checkpoint streams to.public RocksDBStateBackend(StateBackend checkpointStreamBackend, TernaryBoolean enableIncrementalCheckpointing)
RocksDBStateBackend
that uses the given state backend to store its
checkpoint data streams. Typically, one would supply a filesystem or database state backend
here where the snapshots from RocksDB would be stored.
The snapshots of the RocksDB state will be stored using the given backend's
StateBackend.createCheckpointStorage(JobID)
.
checkpointStreamBackend
- The backend write the checkpoint streams to.enableIncrementalCheckpointing
- True if incremental checkpointing is enabled.@Deprecated public RocksDBStateBackend(AbstractStateBackend checkpointStreamBackend)
RocksDBStateBackend(StateBackend)
instead.@Deprecated public RocksDBStateBackend(AbstractStateBackend checkpointStreamBackend, boolean enableIncrementalCheckpointing)
RocksDBStateBackend(StateBackend, TernaryBoolean)
instead.public RocksDBStateBackend configure(Configuration config)
configure
in interface ConfigurableStateBackend
config
- the configurationpublic StateBackend getCheckpointBackend()
This RocksDB state backend only implements the RocksDB specific parts, it relies on the 'CheckpointBackend' to persist the checkpoint and savepoint bytes streams.
public CompletedCheckpointStorageLocation resolveCheckpoint(String pointer) throws IOException
StateBackend
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 IOException
.
resolveCheckpoint
in interface StateBackend
pointer
- The external checkpoint pointer to resolve.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.public CheckpointStorage createCheckpointStorage(JobID jobId) throws IOException
StateBackend
createCheckpointStorage
in interface StateBackend
jobId
- The job to store checkpoint data for.IOException
- Thrown if the checkpoint storage cannot be initialized.public <K> AbstractKeyedStateBackend<K> createKeyedStateBackend(Environment env, JobID jobID, String operatorIdentifier, TypeSerializer<K> keySerializer, int numberOfKeyGroups, KeyGroupRange keyGroupRange, TaskKvStateRegistry kvStateRegistry, TtlTimeProvider ttlTimeProvider) throws IOException
StateBackend
AbstractKeyedStateBackend
that is responsible for holding keyed state
and checkpointing it.
Keyed State is state where each value is bound to a key.
createKeyedStateBackend
in interface StateBackend
createKeyedStateBackend
in class AbstractStateBackend
K
- The type of the keys by which the state is organized.IOException
public OperatorStateBackend createOperatorStateBackend(Environment env, String operatorIdentifier) throws Exception
StateBackend
OperatorStateBackend
that can be used for storing operator state.
Operator state is state that is associated with parallel operator (or function) instances, rather than with keys.
createOperatorStateBackend
in interface StateBackend
createOperatorStateBackend
in class AbstractStateBackend
env
- The runtime environment of the executing task.operatorIdentifier
- The identifier of the operator whose state should be stored.Exception
- This method may forward all exceptions that occur while instantiating the backend.public void setDbStoragePath(String path)
Passing null
to this function restores the default behavior, where the configured
temp directories will be used.
path
- The path where the local RocksDB database files are stored.public void setDbStoragePaths(String... paths)
If nothing is configured, these directories default to the TaskManager's local temporary file directories.
Each distinct state will be stored in one path, but when the state backend creates multiple states, they will store their files on different paths.
Passing null
to this function restores the default behavior, where the configured
temp directories will be used.
paths
- The paths across which the local RocksDB database files will be spread.public String[] getDbStoragePaths()
Under these directories on the TaskManager, RocksDB stores its SST files and metadata files. These directories do not need to be persistent, they can be ephermeral, meaning that they are lost on a machine failure, because state in RocksDB is persisted in checkpoints.
If nothing is configured, these directories default to the TaskManager's local temporary file directories.
public boolean isIncrementalCheckpointsEnabled()
public void setPredefinedOptions(PredefinedOptions options)
If a user-defined options factory is set (via setOptions(OptionsFactory)
),
then the options from the factory are applied on top of the here specified
predefined options.
options
- The options to set (must not be null).public PredefinedOptions getPredefinedOptions()
setPredefinedOptions(PredefinedOptions)
)
are PredefinedOptions.DEFAULT
.
If a user-defined options factory is set (via setOptions(OptionsFactory)
),
then the options from the factory are applied on top of the predefined options.
public void setOptions(OptionsFactory optionsFactory)
Options
for the RocksDB instances.
Because the options are not serializable and hold native code references,
they must be specified through a factory.
The options created by the factory here are applied on top of the pre-defined
options profile selected via setPredefinedOptions(PredefinedOptions)
.
If the pre-defined options profile is the default
(PredefinedOptions.DEFAULT
), then the factory fully controls the RocksDB
options.
optionsFactory
- The options factory that lazily creates the RocksDB options.public OptionsFactory getOptions()
public org.rocksdb.DBOptions getDbOptions()
DBOptions
to be used for all RocksDB instances.public org.rocksdb.ColumnFamilyOptions getColumnOptions()
ColumnFamilyOptions
to be used for all RocksDB instances.Copyright © 2014–2020 The Apache Software Foundation. All rights reserved.