• Keyed State
  • Operator State

CheckpointedFunction

  2.     /**
  3.      * This method is called when a snapshot for a checkpoint is requested. This acts as a hook to the function to
  4.      * ensure that all state is exposed by means previously offered through {@link FunctionInitializationContext} when
  5.      * the Function was initialized, or offered now by {@link FunctionSnapshotContext} itself.
  6.      *
  7.      * @param context the context for drawing a snapshot of the operator
  8.      * @throws Exception
  9.      */
  10.     void snapshotState(FunctionSnapshotContext context) throws Exception;
  12.     /**
  13.      * This method is called when the parallel function instance is created during distributed
  14.      * execution. Functions typically set up their state storing data structures in this method.
  15.      *
  16.      * @param context the context for initializing the operator
  17.      * @throws Exception
  18.      */
  19.     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

  1. public interface FunctionSnapshotContext extends ManagedSnapshotContext {
  2. }

  • FunctionSnapshotContext继承了ManagedSnapshotContext接口

    ManagedSnapshotContext

    1. public interface ManagedSnapshotContext {
    3.   /**
    4.    * Returns the ID of the checkpoint for which the snapshot is taken.
    5.    * 
    6.    * <p>The checkpoint ID is guaranteed to be strictly monotonously increasing across checkpoints.
    7.    * For two completed checkpoints <i>A</i> and <i>B</i>, {@code ID_B > ID_A} means that checkpoint
    8.    * <i>B</i> subsumes checkpoint <i>A</i>, i.e., checkpoint <i>B</i> contains a later state
    9.    * than checkpoint <i>A</i>.
    10.    */
    11.   long getCheckpointId();
    13.   /**
    14.    * Returns timestamp (wall clock time) when the master node triggered the checkpoint for which
    15.    * the state snapshot is taken.
    16.    */
    17.   long getCheckpointTimestamp();
    18. }

  • ManagedSnapshotContext定义了getCheckpointId、getCheckpointTimestamp方法

    FunctionInitializationContext

    1. public interface FunctionInitializationContext extends ManagedInitializationContext {
    2. }

  • FunctionInitializationContext继承了ManagedInitializationContext接口

    ManagedInitializationContext

    ```java 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中的一个实例绑定 (一个 Task 对应一个 state)
    • 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对象

参考