org.apache.flink.contrib.streaming.state

Class DbStateBackend

java.lang.Object

org.apache.flink.runtime.state.AbstractStateBackend

org.apache.flink.contrib.streaming.state.DbStateBackend

All Implemented Interfaces:

Serializable

public class DbStateBackend
extends AbstractStateBackend

AbstractStateBackend for storing checkpoints in JDBC supporting databases. Key-Value state is stored out-of-core and is lazily fetched using the LazyDbValueState implementation. A different backend can also be provided in the constructor to store the non-partitioned states. A common use case would be to store the key-value states in the database and store larger non-partitioned states on a distributed file system.

This backend implementation also allows the sharding of the checkpointed states among multiple database instances, which can be enabled by passing multiple database urls to the DbBackendConfig instance.

By default there are multiple tables created in the given databases: 1 table for non-partitioned checkpoints and 1 table for each key-value state in the streaming program.

To control table creation, insert/lookup operations and to provide compatibility for different SQL implementations, a custom MySqlAdapter can be supplied in the DbBackendConfig.

See Also:

Serialized Form

Nested Class Summary

Nested classes/interfaces inherited from class org.apache.flink.runtime.state.AbstractStateBackend

AbstractStateBackend.CheckpointStateOutputStream, AbstractStateBackend.CheckpointStateOutputView

Field Summary

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

currentKey, keySerializer, userCodeClassLoader

Constructor Summary

Constructors

Constructor and Description
DbStateBackend(DbBackendConfig backendConfig) Create a new DbStateBackend using the provided DbBackendConfig configuration.
DbStateBackend(DbBackendConfig backendConfig, AbstractStateBackend backend) Create a new DbStateBackend using the provided DbBackendConfig configuration and a different backend for storing non-partitioned state snapshots.

Method Summary

Modifier and Type	Method and Description
<S extends Serializable> StateHandle<S>	checkpointStateSerializable(S state, long checkpointID, long timestamp) Writes the given state into the checkpoint, and returns a handle that can retrieve the state back.
void	close() Closes the state backend, releasing all internal resources, but does not delete any persistent checkpoint data.
AbstractStateBackend.CheckpointStateOutputStream	createCheckpointStateOutputStream(long checkpointID, long timestamp) Creates an output stream that writes into the state of the given checkpoint.
protected <N,T,ACC> FoldingState<T,ACC>	createFoldingState(TypeSerializer<N> namespaceSerializer, FoldingStateDescriptor<T,ACC> stateDesc) Creates and returns a new FoldingState.
protected <N,T> ListState<T>	createListState(TypeSerializer<N> namespaceSerializer, ListStateDescriptor<T> stateDesc) Creates and returns a new ListState.
protected <N,T> ReducingState<T>	createReducingState(TypeSerializer<N> namespaceSerializer, ReducingStateDescriptor<T> stateDesc) Creates and returns a new ReducingState.
protected <N,T> ValueState<T>	createValueState(TypeSerializer<N> namespaceSerializer, ValueStateDescriptor<T> stateDesc) Creates and returns a new ValueState.
void	disposeAllStateForCurrentJob() Disposes all state associated with the current job.
DbBackendConfig	getConfiguration() Get the backend configuration object.
ShardedConnection	getConnections() Get the database connections maintained by the backend.
Environment	getEnvironment()
void	initializeForJob(Environment env, String operatorIdentifier, TypeSerializer<?> keySerializer) This method is called by the task upon deployment to initialize the state backend for data for a specific job.
boolean	isInitialized() Check whether the backend has been initialized.

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

createCheckpointStateOutputView, dispose, getCurrentKey, getPartitionedState, injectKeyValueStateSnapshots, notifyOfCompletedCheckpoint, setCurrentKey, snapshotPartitionedState

Methods inherited from class java.lang.Object

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

Constructor Detail

DbStateBackend

public DbStateBackend(DbBackendConfig backendConfig)

Create a new DbStateBackend using the provided DbBackendConfig configuration.

DbStateBackend

public DbStateBackend(DbBackendConfig backendConfig,
                      AbstractStateBackend backend)

Create a new DbStateBackend using the provided DbBackendConfig configuration and a different backend for storing non-partitioned state snapshots.

Method Detail

getConnections

public ShardedConnection getConnections()

Get the database connections maintained by the backend.

isInitialized

public boolean isInitialized()

Check whether the backend has been initialized.

getEnvironment

public Environment getEnvironment()

getConfiguration

public DbBackendConfig getConfiguration()

Get the backend configuration object.

checkpointStateSerializable

public <S extends Serializable> StateHandle<S> checkpointStateSerializable(S state,
                                                                           long checkpointID,
                                                                           long timestamp)
                                                                    throws Exception

Description copied from class: AbstractStateBackend

Writes the given state into the checkpoint, and returns a handle that can retrieve the state back.

Specified by:

checkpointStateSerializable in class AbstractStateBackend

Type Parameters:

S - The type of the state.

Parameters:

state - The state to be checkpointed.

checkpointID - The ID of the checkpoint.

timestamp - The timestamp of the checkpoint.

Returns:

A state handle that can retrieve the checkpoined state.

Throws:

Exception - Exceptions may occur during serialization / storing the state and should be forwarded.

createCheckpointStateOutputStream

public AbstractStateBackend.CheckpointStateOutputStream createCheckpointStateOutputStream(long checkpointID,
                                                                                          long timestamp)
                                                                                   throws Exception

Description copied from class: AbstractStateBackend

Creates an output stream that writes into the state of the given checkpoint. When the stream is closes, it returns a state handle that can retrieve the state back.

Specified by:

createCheckpointStateOutputStream in class AbstractStateBackend

Parameters:

checkpointID - The ID of the checkpoint.

timestamp - The timestamp of the checkpoint.

Returns:

An output stream that writes state for the given checkpoint.

Throws:

Exception - Exceptions may occur while creating the stream and should be forwarded.

createValueState

protected <N,T> ValueState<T> createValueState(TypeSerializer<N> namespaceSerializer,
                                               ValueStateDescriptor<T> stateDesc)
                                        throws Exception

Description copied from class: AbstractStateBackend

Creates and returns a new ValueState.

Specified by:

createValueState in class AbstractStateBackend

Type Parameters:

N - The type of the namespace.

T - The type of the value that the ValueState can store.

Parameters:

namespaceSerializer - TypeSerializer for the state namespace.

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

Throws:

Exception

createListState

protected <N,T> ListState<T> createListState(TypeSerializer<N> namespaceSerializer,
                                             ListStateDescriptor<T> stateDesc)
                                      throws Exception

Description copied from class: AbstractStateBackend

Creates and returns a new ListState.

Specified by:

createListState in class AbstractStateBackend

Type Parameters:

N - The type of the namespace.

T - The type of the values that the ListState can store.

Parameters:

namespaceSerializer - TypeSerializer for the state namespace.

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

Throws:

Exception

createReducingState

protected <N,T> ReducingState<T> createReducingState(TypeSerializer<N> namespaceSerializer,
                                                     ReducingStateDescriptor<T> stateDesc)
                                              throws Exception

Description copied from class: AbstractStateBackend

Creates and returns a new ReducingState.

Specified by:

createReducingState in class AbstractStateBackend

Type Parameters:

N - The type of the namespace.

T - The type of the values that the ListState can store.

Parameters:

namespaceSerializer - TypeSerializer for the state namespace.

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

Throws:

Exception

createFoldingState

protected <N,T,ACC> FoldingState<T,ACC> createFoldingState(TypeSerializer<N> namespaceSerializer,
                                                           FoldingStateDescriptor<T,ACC> stateDesc)
                                                    throws Exception

Description copied from class: AbstractStateBackend

Creates and returns a new FoldingState.

Specified by:

createFoldingState in class AbstractStateBackend

Type Parameters:

N - The type of the namespace.

T - Type of the values folded into the state

ACC - Type of the value in the state *

Parameters:

namespaceSerializer - TypeSerializer for the state namespace.

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

Throws:

Exception

initializeForJob

public void initializeForJob(Environment env,
                             String operatorIdentifier,
                             TypeSerializer<?> keySerializer)
                      throws Exception

Description copied from class: AbstractStateBackend

This method is called by the task upon deployment to initialize the state backend for data for a specific job.

Overrides:

initializeForJob in class AbstractStateBackend

Parameters:

env - The Environment of the task that instantiated the state backend

operatorIdentifier - Unique identifier for naming states created by this backend

Throws:

Exception - Overwritten versions of this method may throw exceptions, in which case the job that uses the state backend is considered failed during deployment.

close

public void close()
           throws Exception

Description copied from class: AbstractStateBackend

Closes the state backend, releasing all internal resources, but does not delete any persistent checkpoint data.

Specified by:

close in class AbstractStateBackend

Throws:

Exception - Exceptions can be forwarded and will be logged by the system

disposeAllStateForCurrentJob

public void disposeAllStateForCurrentJob()
                                  throws Exception

Description copied from class: AbstractStateBackend

Disposes all state associated with the current job.

Specified by:

disposeAllStateForCurrentJob in class AbstractStateBackend

Throws:

Exception - Exceptions may occur during disposal of the state and should be forwarded.