org.apache.flink.contrib.streaming.state

Class RocksDBKeyedStateBackend<K>

java.lang.Object

org.apache.flink.runtime.state.AbstractKeyedStateBackend<K>

org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend<K>

All Implemented Interfaces:

Closeable, AutoCloseable, CheckpointListener, KeyedStateBackend<K>, KeyedStateFactory, PriorityQueueSetFactory, SnapshotStrategy<SnapshotResult<KeyedStateHandle>>, Disposable

public class RocksDBKeyedStateBackend<K>
extends AbstractKeyedStateBackend<K>

An AbstractKeyedStateBackend that stores its state in RocksDB and serializes state to streams provided by a CheckpointStreamFactory upon checkpointing. This state backend can store very large state that exceeds memory and spills to disk. Except for the snapshotting, this class should be accessed as if it is not threadsafe.

This class follows the rules for closing/releasing native RocksDB resources as described in + this document.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static class	RocksDBKeyedStateBackend.RocksDbKvStateInfo Rocks DB specific information about the k/v states.

Nested classes/interfaces inherited from interface org.apache.flink.runtime.state.KeyedStateBackend

KeyedStateBackend.KeySelectionListener<K>

Field Summary

Fields

Modifier and Type	Field and Description
protected org.rocksdb.RocksDB	db Our RocksDB database, this is used by the actual subclasses of AbstractRocksDBState to store state.
static String	MERGE_OPERATOR_NAME The name of the merge operator in RocksDB.

Fields inherited from class org.apache.flink.runtime.state.AbstractKeyedStateBackend

cancelStreamRegistry, keyContext, keyGroupCompressionDecorator, keyGroupRange, keySerializer, kvStateRegistry, numberOfKeyGroups, ttlTimeProvider, userCodeClassLoader

Constructor Summary

Constructors

Constructor and Description

RocksDBKeyedStateBackend(ClassLoader userCodeClassLoader, File instanceBasePath, RocksDBResourceContainer optionsContainer, java.util.function.Function<String,org.rocksdb.ColumnFamilyOptions> columnFamilyOptionsFactory, TaskKvStateRegistry kvStateRegistry, TypeSerializer<K> keySerializer, ExecutionConfig executionConfig, TtlTimeProvider ttlTimeProvider, org.rocksdb.RocksDB db, LinkedHashMap<String,RocksDBKeyedStateBackend.RocksDbKvStateInfo> kvStateInformation, int keyGroupPrefixBytes, CloseableRegistry cancelStreamRegistry, StreamCompressionDecorator keyGroupCompressionDecorator, ResourceGuard rocksDBResourceGuard, RocksDBSnapshotStrategyBase<K> checkpointSnapshotStrategy, RocksDBSnapshotStrategyBase<K> savepointSnapshotStrategy, RocksDBWriteBatchWrapper writeBatchWrapper, org.rocksdb.ColumnFamilyHandle defaultColumnFamilyHandle, RocksDBNativeMetricMonitor nativeMetricMonitor, org.apache.flink.contrib.streaming.state.RocksDBSerializedCompositeKeyBuilder<K> sharedRocksKeyBuilder, PriorityQueueSetFactory priorityQueueFactory, RocksDbTtlCompactFiltersManager ttlCompactFiltersManager, InternalKeyContext<K> keyContext, long writeBatchSize)

Method Summary

Modifier and Type	Method and Description
void	compactState(StateDescriptor<?,?> stateDesc)
<T extends HeapPriorityQueueElement & PriorityComparable & Keyed> KeyGroupedInternalPriorityQueue<T>	create(String stateName, TypeSerializer<T> byteOrderedElementSerializer) Creates a KeyGroupedInternalPriorityQueue.
<N,SV,SEV,S extends State,IS extends S> IS	createInternalState(TypeSerializer<N> namespaceSerializer, StateDescriptor<S,SV> stateDesc, StateSnapshotTransformer.StateSnapshotTransformFactory<SEV> snapshotTransformFactory) Creates and returns a new InternalKvState.
void	dispose() Should only be called by one thread, and only after all accesses to the DB happened.
int	getKeyGroupPrefixBytes()
<N> java.util.stream.Stream<K>	getKeys(String state, N namespace)
org.rocksdb.ReadOptions	getReadOptions()
org.rocksdb.WriteOptions	getWriteOptions()
void	notifyCheckpointAborted(long checkpointId) This method is called as a notification once a distributed checkpoint has been aborted.
void	notifyCheckpointComplete(long completedCheckpointId) This method is called as a notification once a distributed checkpoint has been completed.
int	numKeyValueStateEntries() Returns the total number of state entries across all keys/namespaces.
boolean	requiresLegacySynchronousTimerSnapshots()
void	setCurrentKey(K newKey) Sets the current key that is used for partitioned state.
RunnableFuture<SnapshotResult<KeyedStateHandle>>	snapshot(long checkpointId, long timestamp, CheckpointStreamFactory streamFactory, CheckpointOptions checkpointOptions) Triggers an asynchronous snapshot of the keyed state backend from RocksDB.
boolean	supportsAsynchronousSnapshots()

Methods inherited from class org.apache.flink.runtime.state.AbstractKeyedStateBackend

applyToAllKeys, close, deregisterKeySelectionListener, getCurrentKey, getCurrentKeyGroupIndex, getKeyGroupCompressionDecorator, getKeyGroupRange, getKeySerializer, getNumberOfKeyGroups, getOrCreateKeyedState, getPartitionedState, numKeyValueStatesByName, registerKeySelectionListener

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.apache.flink.runtime.state.KeyedStateFactory

createInternalState

Field Detail

MERGE_OPERATOR_NAME

public static final String MERGE_OPERATOR_NAME

The name of the merge operator in RocksDB. Do not change except you know exactly what you do.

See Also:

Constant Field Values

db

protected final org.rocksdb.RocksDB db

Our RocksDB database, this is used by the actual subclasses of AbstractRocksDBState to store state. The different k/v states that we have don't each have their own RocksDB instance. They all write to this instance but to their own column family.

Constructor Detail

RocksDBKeyedStateBackend

public RocksDBKeyedStateBackend(ClassLoader userCodeClassLoader,
                                File instanceBasePath,
                                RocksDBResourceContainer optionsContainer,
                                java.util.function.Function<String,org.rocksdb.ColumnFamilyOptions> columnFamilyOptionsFactory,
                                TaskKvStateRegistry kvStateRegistry,
                                TypeSerializer<K> keySerializer,
                                ExecutionConfig executionConfig,
                                TtlTimeProvider ttlTimeProvider,
                                org.rocksdb.RocksDB db,
                                LinkedHashMap<String,RocksDBKeyedStateBackend.RocksDbKvStateInfo> kvStateInformation,
                                int keyGroupPrefixBytes,
                                CloseableRegistry cancelStreamRegistry,
                                StreamCompressionDecorator keyGroupCompressionDecorator,
                                ResourceGuard rocksDBResourceGuard,
                                RocksDBSnapshotStrategyBase<K> checkpointSnapshotStrategy,
                                RocksDBSnapshotStrategyBase<K> savepointSnapshotStrategy,
                                RocksDBWriteBatchWrapper writeBatchWrapper,
                                org.rocksdb.ColumnFamilyHandle defaultColumnFamilyHandle,
                                RocksDBNativeMetricMonitor nativeMetricMonitor,
                                org.apache.flink.contrib.streaming.state.RocksDBSerializedCompositeKeyBuilder<K> sharedRocksKeyBuilder,
                                PriorityQueueSetFactory priorityQueueFactory,
                                RocksDbTtlCompactFiltersManager ttlCompactFiltersManager,
                                InternalKeyContext<K> keyContext,
                                @Nonnegative
                                long writeBatchSize)

Method Detail

getKeys

public <N> java.util.stream.Stream<K> getKeys(String state,
                                              N namespace)

Parameters:

state - State variable for which existing keys will be returned.

namespace - Namespace for which existing keys will be returned.

Returns:

A stream of all keys for the given state and namespace. Modifications to the state during iterating over it keys are not supported.

setCurrentKey

public void setCurrentKey(K newKey)

Description copied from interface: KeyedStateBackend

Sets the current key that is used for partitioned state.

Specified by:

setCurrentKey in interface KeyedStateBackend<K>

Overrides:

setCurrentKey in class AbstractKeyedStateBackend<K>

Parameters:

newKey - The new current key.

See Also:

KeyedStateBackend

dispose

public void dispose()

Should only be called by one thread, and only after all accesses to the DB happened.

Specified by:

dispose in interface KeyedStateBackend<K>

Specified by:

dispose in interface Disposable

Overrides:

dispose in class AbstractKeyedStateBackend<K>

create

@Nonnull
public <T extends HeapPriorityQueueElement & PriorityComparable & Keyed> KeyGroupedInternalPriorityQueue<T> create(@Nonnull
                                                                                                                            String stateName,
                                                                                                                            @Nonnull
                                                                                                                            TypeSerializer<T> byteOrderedElementSerializer)

Description copied from interface: PriorityQueueSetFactory

Creates a KeyGroupedInternalPriorityQueue.

Type Parameters:

T - type of the stored elements.

Parameters:

stateName - unique name for associated with this queue.

byteOrderedElementSerializer - a serializer that with a format that is lexicographically ordered in alignment with elementPriorityComparator.

Returns:

the queue with the specified unique name.

getKeyGroupPrefixBytes

public int getKeyGroupPrefixBytes()

getWriteOptions

public org.rocksdb.WriteOptions getWriteOptions()

getReadOptions

public org.rocksdb.ReadOptions getReadOptions()

snapshot

@Nonnull
public RunnableFuture<SnapshotResult<KeyedStateHandle>> snapshot(long checkpointId,
                                                                          long timestamp,
                                                                          @Nonnull
                                                                          CheckpointStreamFactory streamFactory,
                                                                          @Nonnull
                                                                          CheckpointOptions checkpointOptions)
                                                                   throws Exception

Triggers an asynchronous snapshot of the keyed state backend from RocksDB. This snapshot can be canceled and is also stopped when the backend is closed through dispose(). For each backend, this method must always be called by the same thread.

Parameters:

checkpointId - The Id of the checkpoint.

timestamp - The timestamp of the checkpoint.

streamFactory - The factory that we can use for writing our state to streams.

checkpointOptions - Options for how to perform this checkpoint.

Returns:

Future to the state handle of the snapshot data.

Throws:

Exception - indicating a problem in the synchronous part of the checkpoint.

notifyCheckpointComplete

public void notifyCheckpointComplete(long completedCheckpointId)
                              throws Exception

Description copied from interface: CheckpointListener

This method is called as a notification once a distributed checkpoint has been completed.

Note that any exception during this method will not cause the checkpoint to fail any more.

Parameters:

completedCheckpointId - The ID of the checkpoint that has been completed.

Throws:

Exception - This method can propagate exceptions, which leads to a failure/recovery for the task. Not that this will NOT lead to the checkpoint being revoked.

notifyCheckpointAborted

public void notifyCheckpointAborted(long checkpointId)
                             throws Exception

Description copied from interface: CheckpointListener

This method is called as a notification once a distributed checkpoint has been aborted.

Parameters:

checkpointId - The ID of the checkpoint that has been aborted.

Throws:

Exception - This method can propagate exceptions, which leads to a failure/recovery for the task.

createInternalState

@Nonnull
public <N,SV,SEV,S extends State,IS extends S> IS createInternalState(@Nonnull
                                                                               TypeSerializer<N> namespaceSerializer,
                                                                               @Nonnull
                                                                               StateDescriptor<S,SV> stateDesc,
                                                                               @Nonnull
                                                                               StateSnapshotTransformer.StateSnapshotTransformFactory<SEV> snapshotTransformFactory)
                                                                        throws Exception

Description copied from interface: KeyedStateFactory

Creates and returns a new InternalKvState.

Type Parameters:

N - The type of the namespace.

SV - The type of the stored state value.

SEV - The type of the stored state value or entry for collection types (list or map).

S - The type of the public API state.

IS - The type of internal state.

Parameters:

namespaceSerializer - TypeSerializer for the state namespace.

stateDesc - The StateDescriptor that contains the name of the state.

snapshotTransformFactory - factory of state snapshot transformer.

Throws:

Exception

supportsAsynchronousSnapshots

public boolean supportsAsynchronousSnapshots()

Overrides:

supportsAsynchronousSnapshots in class AbstractKeyedStateBackend<K>

numKeyValueStateEntries

@VisibleForTesting
public int numKeyValueStateEntries()

Description copied from class: AbstractKeyedStateBackend

Returns the total number of state entries across all keys/namespaces.

Specified by:

numKeyValueStateEntries in class AbstractKeyedStateBackend<K>

requiresLegacySynchronousTimerSnapshots

public boolean requiresLegacySynchronousTimerSnapshots()

Overrides:

requiresLegacySynchronousTimerSnapshots in class AbstractKeyedStateBackend<K>

compactState

@VisibleForTesting
public void compactState(StateDescriptor<?,?> stateDesc)
                                     throws org.rocksdb.RocksDBException

Throws:

org.rocksdb.RocksDBException