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
setRocksDBOptions(RocksDBOptionsFactory)
.
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,
ClassLoader classLoader)
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,
MetricGroup metricGroup,
Collection<KeyedStateHandle> stateHandles,
CloseableRegistry cancelStreamRegistry)
Creates a new
AbstractKeyedStateBackend that is responsible for holding keyed state
and checkpointing it. |
OperatorStateBackend |
createOperatorStateBackend(Environment env,
String operatorIdentifier,
Collection<OperatorStateHandle> stateHandles,
CloseableRegistry cancelStreamRegistry)
Creates a new
OperatorStateBackend that can be used for storing operator state. |
void |
disableTtlCompactionFilter()
Deprecated.
enabled by default and will be removed in the future.
|
void |
enableTtlCompactionFilter()
Deprecated.
enabled by default and will be removed in the future.
|
StateBackend |
getCheckpointBackend()
Gets the state backend that this RocksDB state backend uses to persist
its bytes to.
|
String[] |
getDbStoragePaths()
Gets the configured local DB storage paths, or null, if none were configured.
|
RocksDBMemoryConfiguration |
getMemoryConfiguration()
Gets the memory configuration object, which offers settings to control RocksDB's memory usage.
|
int |
getNumberOfTransferingThreads()
Deprecated.
Typo in method name. Use
getNumberOfTransferThreads() instead. |
int |
getNumberOfTransferThreads()
Gets the number of threads used to transfer files while snapshotting/restoring.
|
OptionsFactory |
getOptions()
Deprecated.
Use
setRocksDBOptions(RocksDBOptionsFactory) and getRocksDBOptions() instead. |
PredefinedOptions |
getPredefinedOptions()
Gets the currently set predefined options for RocksDB.
|
RocksDBStateBackend.PriorityQueueStateType |
getPriorityQueueStateType()
Gets the type of the priority queue state.
|
RocksDBOptionsFactory |
getRocksDBOptions()
Gets
Options for the RocksDB instances. |
long |
getWriteBatchSize()
Gets the max batch size will be used in
RocksDBWriteBatchWrapper . |
boolean |
isIncrementalCheckpointsEnabled()
Gets whether incremental checkpoints are enabled for this state backend.
|
boolean |
isTtlCompactionFilterEnabled()
Deprecated.
enabled by default and will be removed in the future.
|
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 |
setNumberOfTransferingThreads(int numberOfTransferingThreads)
Deprecated.
Typo in method name. Use
setNumberOfTransferThreads(int) instead. |
void |
setNumberOfTransferThreads(int numberOfTransferThreads)
Sets the number of threads used to transfer files while snapshotting/restoring.
|
void |
setOptions(OptionsFactory optionsFactory)
Deprecated.
Use
setRocksDBOptions(RocksDBOptionsFactory) instead. |
void |
setPredefinedOptions(PredefinedOptions options)
Sets the predefined options for RocksDB.
|
void |
setPriorityQueueStateType(RocksDBStateBackend.PriorityQueueStateType priorityQueueStateType)
Sets the type of the priority queue state.
|
void |
setRocksDBOptions(RocksDBOptionsFactory optionsFactory)
Sets
Options for the RocksDB instances. |
void |
setWriteBatchSize(long writeBatchSize)
Sets the max batch size will be used in
RocksDBWriteBatchWrapper ,
no positive value will disable memory size controller, just use item count controller. |
String |
toString() |
getCompressionDecorator
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, ClassLoader classLoader)
configure
in interface ConfigurableStateBackend
config
- The configuration.classLoader
- The class loader.public 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, MetricGroup metricGroup, @Nonnull Collection<KeyedStateHandle> stateHandles, CloseableRegistry cancelStreamRegistry) 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.env
- The environment of the task.jobID
- The ID of the job that the task belongs to.operatorIdentifier
- The identifier text of the operator.keySerializer
- The key-serializer for the operator.numberOfKeyGroups
- The number of key-groups aka max parallelism.keyGroupRange
- Range of key-groups for which the to-be-created backend is responsible.kvStateRegistry
- KvStateRegistry helper for this task.ttlTimeProvider
- Provider for TTL logic to judge about state expiration.metricGroup
- The parent metric group for all state backend metrics.stateHandles
- The state handles for restore.cancelStreamRegistry
- The registry to which created closeable objects will be registered during restore.IOException
public OperatorStateBackend createOperatorStateBackend(Environment env, String operatorIdentifier, @Nonnull Collection<OperatorStateHandle> stateHandles, CloseableRegistry cancelStreamRegistry) 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.stateHandles
- The state handles for restore.cancelStreamRegistry
- The registry to register streams to close if task canceled.Exception
- This method may forward all exceptions that occur while instantiating the backend.public RocksDBMemoryConfiguration getMemoryConfiguration()
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()
@Deprecated public boolean isTtlCompactionFilterEnabled()
@Deprecated public void enableTtlCompactionFilter()
Note: User can still decide in state TTL configuration in state descriptor whether the filter is active for particular state or not.
@Deprecated public void disableTtlCompactionFilter()
Note: This is an advanced option and the method should only be used when experiencing serious performance degradations during compaction in RocksDB.
public RocksDBStateBackend.PriorityQueueStateType getPriorityQueueStateType()
public void setPriorityQueueStateType(RocksDBStateBackend.PriorityQueueStateType priorityQueueStateType)
public void setPredefinedOptions(@Nonnull PredefinedOptions options)
If user-configured options within RocksDBConfigurableOptions
is set (through flink-conf.yaml)
or a user-defined options factory is set (via setRocksDBOptions(RocksDBOptionsFactory)
),
then the options from the factory are applied on top of the here specified
predefined options and customized options.
options
- The options to set (must not be null).@VisibleForTesting public PredefinedOptions getPredefinedOptions()
setPredefinedOptions(PredefinedOptions)
)
are PredefinedOptions.DEFAULT
.
If user-configured options within RocksDBConfigurableOptions
is set (through flink-conf.yaml)
of a user-defined options factory is set (via setRocksDBOptions(RocksDBOptionsFactory)
),
then the options from the factory are applied on top of the predefined and customized options.
public void setRocksDBOptions(RocksDBOptionsFactory 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.@Nullable public RocksDBOptionsFactory getRocksDBOptions()
Options
for the RocksDB instances.
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.
@Deprecated public void setOptions(OptionsFactory optionsFactory)
setRocksDBOptions(RocksDBOptionsFactory)
instead.@Deprecated public OptionsFactory getOptions()
setRocksDBOptions(RocksDBOptionsFactory)
and getRocksDBOptions()
instead.public int getNumberOfTransferThreads()
public void setNumberOfTransferThreads(int numberOfTransferThreads)
numberOfTransferThreads
- The number of threads used to transfer files while snapshotting/restoring.@Deprecated public int getNumberOfTransferingThreads()
getNumberOfTransferThreads()
instead.@Deprecated public void setNumberOfTransferingThreads(int numberOfTransferingThreads)
setNumberOfTransferThreads(int)
instead.public long getWriteBatchSize()
RocksDBWriteBatchWrapper
.public void setWriteBatchSize(long writeBatchSize)
RocksDBWriteBatchWrapper
,
no positive value will disable memory size controller, just use item count controller.writeBatchSize
- The size will used to be used in RocksDBWriteBatchWrapper
.Copyright © 2014–2020 The Apache Software Foundation. All rights reserved.