Flink中的状态与容错

/**
* This is the core interface for <i>stateful transformation functions</i>, meaning functions
* that maintain state across individual stream records.
* While more lightweight interfaces exist as shortcuts for various types of state, this interface offer the
* greatest flexibility in managing both <i>keyed state</i> and <i>operator state</i>.
*
* <p>The section <a href="#shortcuts">Shortcuts</a> illustrates the common lightweight
* ways to setup stateful functions typically used instead of the full fledged
* abstraction represented by this interface.
*
* <h1>Initialization</h1>
* The {@link CheckpointedFunction#initializeState(FunctionInitializationContext)} is called when
* the parallel instance of the transformation function is created during distributed execution.
* The method gives access to the {@link FunctionInitializationContext} which in turn gives access
* to the to the {@link OperatorStateStore} and {@link KeyedStateStore}.
*
* <p>The {@code OperatorStateStore} and {@code KeyedStateStore} give access to the data structures
* in which state should be stored for Flink to transparently manage and checkpoint it, such as
* {@link org.apache.flink.api.common.state.ValueState} or
* {@link org.apache.flink.api.common.state.ListState}.
*
* <p><b>Note:</b> The {@code KeyedStateStore} can only be used when the transformation supports
* <i>keyed state</i>, i.e., when it is applied on a keyed stream (after a {@code keyBy(...)}).
*
* <h1>Snapshot</h1>
* The {@link CheckpointedFunction#snapshotState(FunctionSnapshotContext)} is called whenever a
* checkpoint takes a state snapshot of the transformation function. Inside this method, functions typically
* make sure that the checkpointed data structures (obtained in the initialization phase) are up
* to date for a snapshot to be taken. The given snapshot context gives access to the metadata
* of the checkpoint.
*
* <p>In addition, functions can use this method as a hook to flush/commit/synchronize with
* external systems.
*
* <h1>Example</h1>
* The code example below illustrates how to use this interface for a function that keeps counts
* of events per key and per parallel partition (parallel instance of the transformation function
* during distributed execution).
* The example also changes of parallelism, which affect the count-per-parallel-partition by
* adding up the counters of partitions that get merged on scale-down. Note that this is a
* toy example, but should illustrate the basic skeleton for a stateful function.
*
* <p><pre>{@code
* public class MyFunction<T> implements MapFunction<T, T>, CheckpointedFunction {
*
* private ReducingState<Long> countPerKey;
* private ListState<Long> countPerPartition;
*
* private long localCount;
*
* public void initializeState(FunctionInitializationContext context) throws Exception {
* // get the state data structure for the per-key state
* countPerKey = context.getKeyedStateStore().getReducingState(
* new ReducingStateDescriptor<>("perKeyCount", new AddFunction<>(), Long.class));
*
* // get the state data structure for the per-partition state
* countPerPartition = context.getOperatorStateStore().getOperatorState(
* new ListStateDescriptor<>("perPartitionCount", Long.class));
*
* // initialize the "local count variable" based on the operator state
* for (Long l : countPerPartition.get()) {
* localCount += l;
* }
* }
*
* public void snapshotState(FunctionSnapshotContext context) throws Exception {
* // the keyed state is always up to date anyways
* // just bring the per-partition state in shape
* countPerPartition.clear();
* countPerPartition.add(localCount);
* }
*
* public T map(T value) throws Exception {
* // update the states
* countPerKey.add(1L);
* localCount++;
*
* return value;
* }
* }
* }</pre>
*
* <hr>
*
* <h1><a name="shortcuts">Shortcuts</a></h1>
* There are various ways that transformation functions can use state without implementing the
* full-fledged {@code CheckpointedFunction} interface:
*
* <h4>Operator State</h4>
* Checkpointing some state that is part of the function object itself is possible in a simpler way
* by directly implementing the {@link ListCheckpointed} interface.
* That mechanism is similar to the previously used {@link Checkpointed} interface.
*
* <h4>Keyed State</h4>
* Access to keyed state is possible via the {@link RuntimeContext}'s methods:
* <pre>{@code
* public class CountPerKeyFunction<T> extends RichMapFunction<T, T> {
*
* private ValueState<Long> count;
*
* public void open(Configuration cfg) throws Exception {
* count = getRuntimeContext().getState(new ValueStateDescriptor<>("myCount", Long.class));
* }
*
* public T map(T value) throws Exception {
* Long current = count.get();
* count.update(current == null ? 1L : current + 1);
*
* return value;
* }
* }
* }</pre>
*
* @see ListCheckpointed
* @see RuntimeContext
*/

/**
* Base class for all streaming tasks. A task is the unit of local processing that is deployed
* and executed by the TaskManagers. Each task runs one or more {@link StreamOperator}s which form
* the Task's operator chain. Operators that are chained together execute synchronously in the
* same thread and hence on the same stream partition. A common case for these chains
* are successive map/flatmap/filter tasks.
*
* <p>The task chain contains one "head" operator and multiple chained operators.
* The StreamTask is specialized for the type of the head operator: one-input and two-input tasks,
* as well as for sources, iteration heads and iteration tails.
*
* <p>The Task class deals with the setup of the streams read by the head operator, and the streams
* produced by the operators at the ends of the operator chain. Note that the chain may fork and
* thus have multiple ends.
*
* <p>The life cycle of the task is set up as follows:
* <pre>{@code
* -- setInitialState -> provides state of all operators in the chain
*
* -- invoke()
* |
* +----> Create basic utils (config, etc) and load the chain of operators
* +----> operators.setup()
* +----> task specific init()
* +----> initialize-operator-states()
* +----> open-operators()
* +----> run()
* +----> close-operators()
* +----> dispose-operators()
* +----> common cleanup
* +----> task specific cleanup()
* }</pre>
*
* <p>The {@code StreamTask} has a lock object called {@code lock}. All calls to methods on a
* {@code StreamOperator} must be synchronized on this lock object to ensure that no methods
* are called concurrently.
*
* @param <OUT>
* @param <OP>
*/

/**
* A <b>State Backend</b> defines how the state of a streaming application is stored and
* checkpointed. Different State Backends store their state in different fashions, and use
* different data structures to hold the state of a running application.
*
* <p>For example, the {@link org.apache.flink.runtime.state.memory.MemoryStateBackend memory state backend}
* keeps working state in the memory of the TaskManager and stores checkpoints in the memory of the
* JobManager. The backend is lightweight and without additional dependencies, but not highly available
* and supports only small state.
*
* <p>The {@link org.apache.flink.runtime.state.filesystem.FsStateBackend file system state backend}
* keeps working state in the memory of the TaskManager and stores state checkpoints in a filesystem
* (typically a replicated highly-available filesystem, like <a href="https://hadoop.apache.org/">HDFS</a>,
* <a href="https://ceph.com/">Ceph</a>, <a href="https://aws.amazon.com/documentation/s3/">S3</a>,
* <a href="https://cloud.google.com/storage/">GCS</a>, etc).
*
* <p>The {@code RocksDBStateBackend} stores working state in <a href="http://rocksdb.org/">RocksDB</a>,
* and checkpoints the state by default to a filesystem (similar to the {@code FsStateBackend}).
*
* <h2>Raw Bytes Storage and Backends</h2>
*
* The {@code StateBackend} creates services for <i>raw bytes storage</i> and for <i>keyed state</i>
* and <i>operator state</i>.
*
* <p>The <i>raw bytes storage</i> (through the {@link CheckpointStreamFactory}) is the fundamental
* service that simply stores bytes in a fault tolerant fashion. This service is used by the JobManager
* to store checkpoint and recovery metadata and is typically also used by the keyed- and operator state
* backends to store checkpointed state.
*
* <p>The {@link AbstractKeyedStateBackend} and {@link OperatorStateBackend} created by this state
* backend define how to hold the working state for keys and operators. They also define how to checkpoint
* that state, frequently using the raw bytes storage (via the {@code CheckpointStreamFactory}).
* However, it is also possible that for example a keyed state backend simply implements the bridge to
* a key/value store, and that it does not need to store anything in the raw byte storage upon a
* checkpoint.
*
* <h2>Serializability</h2>
*
* State Backends need to be {@link java.io.Serializable serializable}, because they distributed
* across parallel processes (for distributed execution) together with the streaming application code.
*
* <p>Because of that, {@code StateBackend} implementations (typically subclasses
* of {@link AbstractStateBackend}) are meant to be like <i>factories</i> that create the proper
* states stores that provide access to the persistent storage and hold the keyed- and operator
* state data structures. That way, the State Backend can be very lightweight (contain only
* configurations) which makes it easier to be serializable.
*
* <h2>Thread Safety</h2>
*
* State backend implementations have to be thread-safe. Multiple threads may be creating
* streams and keyed-/operator state backends concurrently.
*/

/**
* Savepoints are manually-triggered snapshots from which a program can be
* resumed on submission.
*
* <p>In order to allow changes to the savepoint format between Flink versions,
* we allow different savepoint implementations (see subclasses of this
* interface).
*
* <p>Savepoints are serialized via a {@link SavepointSerializer}.
*/

/**
* Client for querying Flink's managed state.
*
* <p>You can mark state as queryable via {@link StateDescriptor#setQueryable(String)}.
* The state instance created from this descriptor will be published for queries when it's
* created on the Task Managers and the location will be reported to the Job Manager.
*
* <p>The client connects to a {@code Client Proxy} running on a given Task Manager. The
* proxy is the entry point of the client to the Flink cluster. It forwards the requests
* of the client to the Job Manager and the required Task Manager, and forwards the final
* response back the client.
*
* <p>The proxy, initially resolves the location of the requested KvState via the JobManager. Resolved
* locations are cached. When the server address of the requested KvState instance is determined, the
* client sends out a request to the server. The returned final answer is then forwarded to the Client.
*/

/**
* Returns a future holding the serialized request result.
*
* @param jobId JobID of the job the queryable state
* belongs to
* @param queryableStateName Name under which the state is queryable
* @param keyHashCode Integer hash code of the key (result of
* a call to {@link Object#hashCode()}
* @param serializedKeyAndNamespace Serialized key and namespace to query
* KvState instance with
* @return Future holding the serialized result
*/
private CompletableFuture<KvStateResponse> getKvState(
final JobID jobId,
final String queryableStateName,
final int keyHashCode,
final byte[] serializedKeyAndNamespace) {
LOG.debug("Sending State Request to {}.", remoteAddress);
try {
KvStateRequest request = new KvStateRequest(jobId, queryableStateName, keyHashCode, serializedKeyAndNamespace);
return client.sendRequest(remoteAddress, request);
} catch (Exception e) {
LOG.error("Unable to send KVStateRequest: ", e);
return FutureUtils.getFailedFuture(e);
}
}

/**
* An interface for the Queryable State Server running on each Task Manager in the cluster.
* This server is responsible for serving requests coming from the {@link KvStateClientProxy
* Queryable State Proxy} and requesting <b>locally</b> stored state.
*/