*

The raw bytes storage (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.

*/

4.Savepoint

Savepoint是Checkpoint的一种特殊实现，底层也是使用Checkpoint的机制。Savepoint是用户以手工命令的方式触发Checkpoint并将结果持久化到指定的存储里，其主要目的是帮助用户在升级和维护集群过程中保存系统的状态数据，避免因停机或者升级邓正常终止应用的操作而导致系统无法恢复到原有的计算状态，而无法实现Exactly-Once的语义保证。

/** * 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}.

*/

5.Querable State

Queryable State，顾名思义，就是可查询的状态，表示这个状态，在流计算的过程中就可以被查询，而不像其他流计算框架，需要存储到外部系统中才能被查询。目前可查询的state主要针对partitionable state，如keyed state等。

简单来说，当用户在job中定义了queryable state之后，就可以在外部，通过QueryableStateClient，通过job id, state name, key来查询所对应的状态的实时的值。

5.1 QueryableStateClient

QueryableStateClient

/** * 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

*/privateCompletableFuture getKvState(

final JobID jobId,

final String queryableStateName,

finalint keyHashCode,

finalbyte[] 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);

}

}

通过组装request，然后使用client发送请求

5.2 KvStateServer

KvStateServer

/** * 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.

*/

6. 总结

为什么要使用状态?

数据之间有关联，需要通过状态满足业务逻辑

为什么要管理状态?

实时计算作业需要7*24运行，需要应对不可靠因素带来的影响

如何选择状态的类型和存储方式?

分析自己的业务场景，比对各方案的利弊，选择合适的，够用即可