序
本文主要研究一下flink的CheckpointedFunction
实例
public class BufferingSink implements SinkFunction>, CheckpointedFunction { private final int threshold; private transient ListState > checkpointedState; private List > bufferedElements; public BufferingSink(int threshold) { this.threshold = threshold; this.bufferedElements = new ArrayList<>(); } @Override public void invoke(Tuple2 value) throws Exception { bufferedElements.add(value); if (bufferedElements.size() == threshold) { for (Tuple2 element: bufferedElements) { // send it to the sink } bufferedElements.clear(); } } @Override public void snapshotState(FunctionSnapshotContext context) throws Exception { checkpointedState.clear(); for (Tuple2 element : bufferedElements) { checkpointedState.add(element); } } @Override public void initializeState(FunctionInitializationContext context) throws Exception { ListStateDescriptor > descriptor = new ListStateDescriptor<>( "buffered-elements", TypeInformation.of(new TypeHint >() {})); checkpointedState = context.getOperatorStateStore().getListState(descriptor); if (context.isRestored()) { for (Tuple2 element : checkpointedState.get()) { bufferedElements.add(element); } } }}复制代码
- 这个BufferingSink实现了CheckpointedFunction接口,它定义了ListState类型的checkpointedState,以及List结构的bufferedElements
- 在invoke方法里头先将value缓存到bufferedElements,缓存个数触发阈值时,执行sink操作,然后清空bufferedElements
- 在snapshotState方法里头对bufferedElements进行snapshot操作,在initializeState先创建ListStateDescriptor,然后通过FunctionInitializationContext.getOperatorStateStore().getListState(descriptor)来获取ListState,之后判断state是否有在前一次execution的snapshot中restored,如果有则将ListState中的数据恢复到bufferedElements
CheckpointedFunction
flink-streaming-java_2.11-1.7.0-sources.jar!/org/apache/flink/streaming/api/checkpoint/CheckpointedFunction.java
@PublicEvolving@SuppressWarnings("deprecation")public interface CheckpointedFunction { /** * This method is called when a snapshot for a checkpoint is requested. This acts as a hook to the function to * ensure that all state is exposed by means previously offered through {@link FunctionInitializationContext} when * the Function was initialized, or offered now by {@link FunctionSnapshotContext} itself. * * @param context the context for drawing a snapshot of the operator * @throws Exception */ void snapshotState(FunctionSnapshotContext context) throws Exception; /** * This method is called when the parallel function instance is created during distributed * execution. Functions typically set up their state storing data structures in this method. * * @param context the context for initializing the operator * @throws Exception */ void initializeState(FunctionInitializationContext context) throws Exception;}复制代码
- CheckpointedFunction是stateful transformation functions的核心接口,用于跨stream维护state
- snapshotState在checkpoint的时候会被调用,用于snapshot state,通常用于flush、commit、synchronize外部系统
- initializeState在parallel function初始化的时候(第一次初始化或者从前一次checkpoint recover的时候)被调用,通常用来初始化state,以及处理state recovery的逻辑
FunctionSnapshotContext
flink-runtime_2.11-1.7.0-sources.jar!/org/apache/flink/runtime/state/FunctionSnapshotContext.java
/** * This interface provides a context in which user functions that use managed state (i.e. state that is managed by state * backends) can participate in a snapshot. As snapshots of the backends themselves are taken by the system, this * interface mainly provides meta information about the checkpoint. */@PublicEvolvingpublic interface FunctionSnapshotContext extends ManagedSnapshotContext {}复制代码
- FunctionSnapshotContext继承了ManagedSnapshotContext接口
ManagedSnapshotContext
flink-runtime_2.11-1.7.0-sources.jar!/org/apache/flink/runtime/state/ManagedSnapshotContext.java
/** * This interface provides a context in which operators that use managed state (i.e. state that is managed by state * backends) can perform a snapshot. As snapshots of the backends themselves are taken by the system, this interface * mainly provides meta information about the checkpoint. */@PublicEvolvingpublic interface ManagedSnapshotContext { /** * Returns the ID of the checkpoint for which the snapshot is taken. * *The checkpoint ID is guaranteed to be strictly monotonously increasing across checkpoints. * For two completed checkpoints A and B, {@code ID_B > ID_A} means that checkpoint * B subsumes checkpoint A, i.e., checkpoint B contains a later state * than checkpoint A. */ long getCheckpointId(); /** * Returns timestamp (wall clock time) when the master node triggered the checkpoint for which * the state snapshot is taken. */ long getCheckpointTimestamp();}复制代码
- ManagedSnapshotContext定义了getCheckpointId、getCheckpointTimestamp方法
FunctionInitializationContext
flink-runtime_2.11-1.7.0-sources.jar!/org/apache/flink/runtime/state/FunctionInitializationContext.java
/** * This interface provides a context in which user functions can initialize by registering to managed state (i.e. state * that is managed by state backends). * ** Operator state is available to all functions, while keyed state is only available for functions after keyBy. * *
* For the purpose of initialization, the context signals if the state is empty or was restored from a previous * execution. * */@PublicEvolvingpublic interface FunctionInitializationContext extends ManagedInitializationContext {}复制代码
- FunctionInitializationContext继承了ManagedInitializationContext接口
ManagedInitializationContext
flink-runtime_2.11-1.7.0-sources.jar!/org/apache/flink/runtime/state/ManagedInitializationContext.java
/** * This interface provides a context in which operators can initialize by registering to managed state (i.e. state that * is managed by state backends). * ** Operator state is available to all operators, while keyed state is only available for operators after keyBy. * *
* For the purpose of initialization, the context signals if the state is empty (new operator) or was restored from * a previous execution of this operator. * */public interface ManagedInitializationContext { /** * Returns true, if state was restored from the snapshot of a previous execution. This returns always false for * stateless tasks. */ boolean isRestored(); /** * Returns an interface that allows for registering operator state with the backend. */ OperatorStateStore getOperatorStateStore(); /** * Returns an interface that allows for registering keyed state with the backend. */ KeyedStateStore getKeyedStateStore();}复制代码
- ManagedInitializationContext接口定义了isRestored、getOperatorStateStore、getKeyedStateStore方法
小结
- flink有两种基本的state,分别是Keyed State以及Operator State(
non-keyed state
);其中Keyed State只能在KeyedStream上的functions及operators上使用;每个operator state会跟parallel operator中的一个实例绑定;Operator State支持parallelism变更时进行redistributing - Keyed State及Operator State都分别有managed及raw两种形式,managed由flink runtime来管理,由runtime负责encode及写入checkpoint;raw形式的state由operators自己管理,flink runtime无法了解该state的数据结构,将其视为raw bytes;所有的datastream function都可以使用managed state,而raw state一般仅限于自己实现operators来使用
- stateful function可以通过CheckpointedFunction接口或者ListCheckpointed接口来使用managed operator state;CheckpointedFunction定义了snapshotState、initializeState两个方法;每当checkpoint执行的时候,snapshotState会被调用;而initializeState方法在每次用户定义的function初始化的时候(
第一次初始化或者从前一次checkpoint recover的时候
)被调用,该方法不仅可以用来初始化state,还可以用于处理state recovery的逻辑 - 对于manageed operator state,目前仅仅支持list-style的形式,即要求state是serializable objects的List结构,方便在rescale的时候进行redistributed;关于redistribution schemes的模式目前有两种,分别是Even-split redistribution(
在restore/redistribution的时候每个operator仅仅得到整个state的sublist
)及Union redistribution(在restore/redistribution的时候每个operator得到整个state的完整list
) - FunctionSnapshotContext继承了ManagedSnapshotContext接口,它定义了getCheckpointId、getCheckpointTimestamp方法;FunctionInitializationContext继承了ManagedInitializationContext接口,它定义了isRestored、getOperatorStateStore、getKeyedStateStore方法,可以用来判断是否是在前一次execution的snapshot中restored,以及获取OperatorStateStore、KeyedStateStore对象