聊聊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中的數據恢復到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.
 */
@PublicEvolving
public 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.
 */
@PublicEvolving
public interface ManagedSnapshotContext {

	/**
	 * Returns the ID of the checkpoint for which the snapshot is taken.
	 * 
	 * <p>The checkpoint ID is guaranteed to be strictly monotonously increasing across checkpoints.
	 * For two completed checkpoints <i>A</i> and <i>B</i>, {@code ID_B > ID_A} means that checkpoint
	 * <i>B</i> subsumes checkpoint <i>A</i>, i.e., checkpoint <i>B</i> contains a later state
	 * than checkpoint <i>A</i>.
	 */
	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).
 *
 * <p>
 * Operator state is available to all functions, while keyed state is only available for functions after keyBy.
 *
 * <p>
 * For the purpose of initialization, the context signals if the state is empty or was restored from a previous
 * execution.
 *
 */
@PublicEvolving
public 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).
 *
 * <p>
 * Operator state is available to all operators, while keyed state is only available for operators after keyBy.
 *
 * <p>
 * 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對象

doc

• Using Managed Operator State