org.apache.flink.contrib.streaming.state

Class MySqlAdapter

java.lang.Object

org.apache.flink.contrib.streaming.state.MySqlAdapter

All Implemented Interfaces:

Serializable, DbAdapter

public class MySqlAdapter
extends Object
implements DbAdapter

Adapter for bridging inconsistencies between the different SQL implementations. The default implementation has been tested to work well with MySQL

See Also:

Serialized Form

Constructor Summary

Constructors

Constructor and Description
MySqlAdapter()

Method Summary

Modifier and Type	Method and Description
void	cleanupFailedCheckpoints(String stateId, Connection con, long checkpointTs, long recoveryTs) Clean up states between the checkpoint and recovery timestamp.
void	compactKvStates(String stateId, Connection con, long lowerId, long upperId) Compact the states between two checkpoint timestamp by only keeping the most recent.
void	createCheckpointsTable(String jobId, Connection con) Initialize tables for storing non-partitioned checkpoints for the given job id and database connection.
void	createKVStateTable(String stateId, Connection con) Initialize the necessary tables for the given stateId.
void	deleteCheckpoint(String jobId, Connection con, long checkpointId, long checkpointTs, long handleId) Remove the given checkpoint from the database.
void	disposeAllStateForJob(String jobId, Connection con) Remove all states for the given JobID, by for instance dropping the entire table.
byte[]	getCheckpoint(String jobId, Connection con, long checkpointId, long checkpointTs, long handleId) Retrieve the serialized checkpoint data from the database.
void	insertBatch(String stateId, DbBackendConfig conf, Connection con, PreparedStatement insertStatement, long checkpointTs, List<Tuple2<byte[],byte[]>> toInsert) Insert a list of Key-Value pairs into the database.
void	keepAlive(Connection con) Execute a simple operation to refresh the current database connection in case no data is written for a longer time period.
byte[]	lookupKey(String stateId, PreparedStatement lookupStatement, byte[] key, long lookupTs) Retrieve the latest value from the database for a given key and timestamp.
PreparedStatement	prepareCheckpointInsert(String jobId, Connection con) Checkpoints will be inserted in the database using prepared statements.
String	prepareKeyLookup(String stateId) Prepare the statement that will be used to lookup keys from the database.
String	prepareKVCheckpointInsert(String stateId) Prepare the the statement that will be used to insert key-value pairs in the database.
void	setCheckpointInsertParams(String jobId, PreparedStatement insertStatement, long checkpointId, long timestamp, long handleId, byte[] checkpoint) Set the PreparedStatement parameters for the statement returned by DbAdapter.prepareCheckpointInsert(String, Connection).
protected static void	validateStateId(String name) Tries to avoid SQL injection with weird state names.

Methods inherited from class java.lang.Object

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

Constructor Detail

MySqlAdapter

public MySqlAdapter()

Method Detail

createCheckpointsTable

public void createCheckpointsTable(String jobId,
                                   Connection con)
                            throws SQLException

Description copied from interface: DbAdapter

Initialize tables for storing non-partitioned checkpoints for the given job id and database connection.

Specified by:

createCheckpointsTable in interface DbAdapter

Throws:

SQLException

prepareCheckpointInsert

public PreparedStatement prepareCheckpointInsert(String jobId,
                                                 Connection con)
                                          throws SQLException

Description copied from interface: DbAdapter

Checkpoints will be inserted in the database using prepared statements. This methods should prepare and return the statement that will be used later to insert using the given connection.

Specified by:

prepareCheckpointInsert in interface DbAdapter

Throws:

SQLException

setCheckpointInsertParams

public void setCheckpointInsertParams(String jobId,
                                      PreparedStatement insertStatement,
                                      long checkpointId,
                                      long timestamp,
                                      long handleId,
                                      byte[] checkpoint)
                               throws SQLException

Description copied from interface: DbAdapter

Set the PreparedStatement parameters for the statement returned by DbAdapter.prepareCheckpointInsert(String, Connection).

Specified by:

setCheckpointInsertParams in interface DbAdapter

Parameters:

jobId - Id of the current job.

insertStatement - Statement returned by DbAdapter.prepareCheckpointInsert(String, Connection).

checkpointId - Global checkpoint id.

timestamp - Global checkpoint timestamp.

handleId - Unique id assigned to this state checkpoint (should be primary key).

checkpoint - The serialized checkpoint.

Throws:

SQLException

getCheckpoint

public byte[] getCheckpoint(String jobId,
                            Connection con,
                            long checkpointId,
                            long checkpointTs,
                            long handleId)
                     throws SQLException

Description copied from interface: DbAdapter

Retrieve the serialized checkpoint data from the database.

Specified by:

getCheckpoint in interface DbAdapter

Parameters:

jobId - Id of the current job.

con - Database connection

checkpointId - Global checkpoint id.

checkpointTs - Global checkpoint timestamp.

handleId - Unique id assigned to this state checkpoint (should be primary key).

Returns:

The byte[] corresponding to the checkpoint or null if missing.

Throws:

SQLException

deleteCheckpoint

public void deleteCheckpoint(String jobId,
                             Connection con,
                             long checkpointId,
                             long checkpointTs,
                             long handleId)
                      throws SQLException

Description copied from interface: DbAdapter

Remove the given checkpoint from the database.

Specified by:

deleteCheckpoint in interface DbAdapter

Parameters:

jobId - Id of the current job.

con - Database connection

checkpointId - Global checkpoint id.

checkpointTs - Global checkpoint timestamp.

handleId - Unique id assigned to this state checkpoint (should be primary key).

Throws:

SQLException

disposeAllStateForJob

public void disposeAllStateForJob(String jobId,
                                  Connection con)
                           throws SQLException

Description copied from interface: DbAdapter

Remove all states for the given JobID, by for instance dropping the entire table.

Specified by:

disposeAllStateForJob in interface DbAdapter

Throws:

SQLException

createKVStateTable

public void createKVStateTable(String stateId,
                               Connection con)
                        throws SQLException

Description copied from interface: DbAdapter

Initialize the necessary tables for the given stateId. The state id consist of the JobId+OperatorId+StateName.

Specified by:

createKVStateTable in interface DbAdapter

Throws:

SQLException

prepareKVCheckpointInsert

public String prepareKVCheckpointInsert(String stateId)
                                 throws SQLException

Description copied from interface: DbAdapter

Prepare the the statement that will be used to insert key-value pairs in the database.

Specified by:

prepareKVCheckpointInsert in interface DbAdapter

Throws:

SQLException

prepareKeyLookup

public String prepareKeyLookup(String stateId)
                        throws SQLException

Description copied from interface: DbAdapter

Prepare the statement that will be used to lookup keys from the database. Keys and values are assumed to be byte arrays.

Specified by:

prepareKeyLookup in interface DbAdapter

Throws:

SQLException

lookupKey

public byte[] lookupKey(String stateId,
                        PreparedStatement lookupStatement,
                        byte[] key,
                        long lookupTs)
                 throws SQLException

Description copied from interface: DbAdapter

Retrieve the latest value from the database for a given key and timestamp.

Specified by:

lookupKey in interface DbAdapter

Parameters:

stateId - Unique identifier of the kvstate (usually the table name).

lookupStatement - The statement returned by #prepareKeyLookup(String, Connection).

key - The key to lookup.

Returns:

The latest valid value for the key.

Throws:

SQLException

cleanupFailedCheckpoints

public void cleanupFailedCheckpoints(String stateId,
                                     Connection con,
                                     long checkpointTs,
                                     long recoveryTs)
                              throws SQLException

Description copied from interface: DbAdapter

Clean up states between the checkpoint and recovery timestamp.

Specified by:

cleanupFailedCheckpoints in interface DbAdapter

Throws:

SQLException

compactKvStates

public void compactKvStates(String stateId,
                            Connection con,
                            long lowerId,
                            long upperId)
                     throws SQLException

Description copied from interface: DbAdapter

Compact the states between two checkpoint timestamp by only keeping the most recent.

Specified by:

compactKvStates in interface DbAdapter

Throws:

SQLException

validateStateId

protected static void validateStateId(String name)

Tries to avoid SQL injection with weird state names.

insertBatch

public void insertBatch(String stateId,
                        DbBackendConfig conf,
                        Connection con,
                        PreparedStatement insertStatement,
                        long checkpointTs,
                        List<Tuple2<byte[],byte[]>> toInsert)
                 throws IOException

Description copied from interface: DbAdapter

Insert a list of Key-Value pairs into the database. The suggested approach is to use idempotent inserts(updates) as 1 batch operation.

Specified by:

insertBatch in interface DbAdapter

Throws:

IOException

keepAlive

public void keepAlive(Connection con)
               throws SQLException

Description copied from interface: DbAdapter

Execute a simple operation to refresh the current database connection in case no data is written for a longer time period. Usually something like "select 1"

Specified by:

keepAlive in interface DbAdapter

Throws:

SQLException