聊聊flink的CheckpointedFunction

序本文主要研究一下flink的CheckpointedFunction实例public class BufferingSink implements SinkFunction<Tuple2<String, Integer>>, CheckpointedFunction { private final int threshold; private transient ListState<Tuple2<String, Integer>> checkpointedState; private List<Tuple2<String, Integer>> bufferedElements; public BufferingSink(int threshold) { this.threshold = threshold; this.bufferedElements = new ArrayList<>(); } @Override public void invoke(Tuple2<String, Integer> value) throws Exception { bufferedElements.add(value); if (bufferedElements.size() == threshold) { for (Tuple2<String, Integer> element: bufferedElements) { // send it to the sink } bufferedElements.clear(); } } @Override public void snapshotState(FunctionSnapshotContext context) throws Exception { checkpointedState.clear(); for (Tuple2<String, Integer> element : bufferedElements) { checkpointedState.add(element); } } @Override public void initializeState(FunctionInitializationContext context) throws Exception { ListStateDescriptor<Tuple2<String, Integer>> descriptor = new ListStateDescriptor<>( “buffered-elements”, TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {})); checkpointedState = context.getOperatorStateStore().getListState(descriptor); if (context.isRestored()) { for (Tuple2<String, Integer> 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中的数据恢复到bufferedElementsCheckpointedFunctionflink-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维护statesnapshotState在checkpoint的时候会被调用，用于snapshot state，通常用于flush、commit、synchronize外部系统initializeState在parallel function被创建时调用，通常用于初始化存储于state的数据FunctionSnapshotContextflink-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接口ManagedSnapshotContextflink-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方法FunctionInitializationContextflink-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接口ManagedInitializationContextflink-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变更时进行redistributingKeyed 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对象docUsing Managed Operator State