org.apache.flink.contrib.streaming.state

Class AbstractRocksDBState<K,N,S extends State,SD extends StateDescriptor<S,?>>

java.lang.Object

org.apache.flink.contrib.streaming.state.AbstractRocksDBState<K,N,S,SD>

Type Parameters:

K - The type of the key.

N - The type of the namespace.

S - The type of State.

SD - The type of StateDescriptor.

All Implemented Interfaces:

State, KvState<K,N,S,SD,RocksDBStateBackend>

Direct Known Subclasses:

RocksDBFoldingState, RocksDBListState, RocksDBReducingState, RocksDBValueState

public abstract class AbstractRocksDBState<K,N,S extends State,SD extends StateDescriptor<S,?>>
extends Object
implements KvState<K,N,S,SD,RocksDBStateBackend>, State

Base class for State implementations that store state in a RocksDB database.

This base class is responsible for setting up the RocksDB database, for checkpointing/restoring the database and for disposal in the dispose() method. The concrete subclasses just use the RocksDB handle to store/retrieve state.

State is checkpointed asynchronously. The synchronous part is drawing the actual backup from RocksDB, this is done in snapshot(long, long). This will return a AsyncRocksDBSnapshot that will perform the copying of the backup to the remote file system.

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
protected static class	AbstractRocksDBState.AbstractRocksDBSnapshot<K,N,S extends State,SD extends StateDescriptor<S,?>>

Field Summary

Fields

Modifier and Type	Field and Description
protected File	basePath Store it so that we can clean up in dispose()
protected String	checkpointPath FileSystem path where checkpoints are stored
protected K	currentKey The current key, which the next value methods will refer to
protected N	currentNamespace The current namespace, which the next value methods will refer to
protected org.rocksdb.RocksDB	db Our RocksDB instance
protected TypeSerializer<K>	keySerializer Serializer for the keys
protected TypeSerializer<N>	namespaceSerializer Serializer for the namespace
protected File	rocksDbPath Directory in "basePath" where the actual RocksDB data base instance stores its files

Constructor Summary

Constructors

Modifier	Constructor and Description
protected	AbstractRocksDBState(TypeSerializer<K> keySerializer, TypeSerializer<N> namespaceSerializer, File basePath, String checkpointPath, org.rocksdb.Options options) Creates a new RocksDB backed state.
protected	AbstractRocksDBState(TypeSerializer<K> keySerializer, TypeSerializer<N> namespaceSerializer, File basePath, String checkpointPath, String restorePath, org.rocksdb.Options options) Creates a new RocksDB backed state and restores from the given backup directory.

Method Summary

Modifier and Type	Method and Description
void	clear() Removes the value mapped under the current key.
protected abstract AbstractRocksDBState.AbstractRocksDBSnapshot<K,N,S,SD>	createRocksDBSnapshot(URI backupUri, long checkpointId)
void	dispose() Disposes the key/value state, releasing all occupied resources.
void	setCurrentKey(K currentKey) Sets the current key, which will be used when using the state access methods.
void	setCurrentNamespace(N namespace) Sets the current namespace, which will be used when using the state access methods.
KvStateSnapshot<K,N,S,SD,RocksDBStateBackend>	snapshot(long checkpointId, long timestamp) Creates a snapshot of this state.
protected void	writeKeyAndNamespace(DataOutputView out)

Methods inherited from class java.lang.Object

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

Field Detail

keySerializer

protected final TypeSerializer<K> keySerializer

Serializer for the keys

namespaceSerializer

protected final TypeSerializer<N> namespaceSerializer

Serializer for the namespace

currentKey

protected K currentKey

The current key, which the next value methods will refer to

currentNamespace

protected N currentNamespace

The current namespace, which the next value methods will refer to

basePath

protected final File basePath

Store it so that we can clean up in dispose()

checkpointPath

protected final String checkpointPath

FileSystem path where checkpoints are stored

rocksDbPath

protected final File rocksDbPath

Directory in "basePath" where the actual RocksDB data base instance stores its files

db

protected final org.rocksdb.RocksDB db

Our RocksDB instance

Constructor Detail

AbstractRocksDBState

protected AbstractRocksDBState(TypeSerializer<K> keySerializer,
                               TypeSerializer<N> namespaceSerializer,
                               File basePath,
                               String checkpointPath,
                               org.rocksdb.Options options)

Creates a new RocksDB backed state.

Parameters:

keySerializer - The serializer for the keys.

namespaceSerializer - The serializer for the namespace.

basePath - The path on the local system where RocksDB data should be stored.

AbstractRocksDBState

protected AbstractRocksDBState(TypeSerializer<K> keySerializer,
                               TypeSerializer<N> namespaceSerializer,
                               File basePath,
                               String checkpointPath,
                               String restorePath,
                               org.rocksdb.Options options)

Creates a new RocksDB backed state and restores from the given backup directory. After restoring the backup directory is deleted.

Parameters:

keySerializer - The serializer for the keys.

namespaceSerializer - The serializer for the namespace.

basePath - The path on the local system where RocksDB data should be stored.

restorePath - The path to a backup directory from which to restore RocksDb database.

Method Detail

clear

public final void clear()

Description copied from interface: State

Removes the value mapped under the current key.

Specified by:

clear in interface State

writeKeyAndNamespace

protected void writeKeyAndNamespace(DataOutputView out)
                             throws IOException

Throws:

IOException

setCurrentKey

public final void setCurrentKey(K currentKey)

Description copied from interface: KvState

Sets the current key, which will be used when using the state access methods.

Specified by:

setCurrentKey in interface KvState<K,N,S extends State,SD extends StateDescriptor<S,?>,RocksDBStateBackend>

Parameters:

currentKey - The key.

setCurrentNamespace

public final void setCurrentNamespace(N namespace)

Description copied from interface: KvState

Sets the current namespace, which will be used when using the state access methods.

Specified by:

setCurrentNamespace in interface KvState<K,N,S extends State,SD extends StateDescriptor<S,?>,RocksDBStateBackend>

Parameters:

namespace - The namespace.

createRocksDBSnapshot

protected abstract AbstractRocksDBState.AbstractRocksDBSnapshot<K,N,S,SD> createRocksDBSnapshot(URI backupUri,
                                                                                                long checkpointId)

snapshot

public final KvStateSnapshot<K,N,S,SD,RocksDBStateBackend> snapshot(long checkpointId,
                                                                    long timestamp)
                                                             throws Exception

Description copied from interface: KvState

Creates a snapshot of this state.

Specified by:

snapshot in interface KvState<K,N,S extends State,SD extends StateDescriptor<S,?>,RocksDBStateBackend>

Parameters:

checkpointId - The ID of the checkpoint for which the snapshot should be created.

timestamp - The timestamp of the checkpoint.

Returns:

A snapshot handle for this key/value state.

Throws:

Exception - Exceptions during snapshotting the state should be forwarded, so the system can react to failed snapshots.

dispose

public final void dispose()

Description copied from interface: KvState

Disposes the key/value state, releasing all occupied resources.

Specified by:

dispose in interface KvState<K,N,S extends State,SD extends StateDescriptor<S,?>,RocksDBStateBackend>