聊聊flink的ListCheckpointed

序

本文主要研究一下flink的ListCheckpointed

實例

public static class CounterSource
        extends RichParallelSourceFunction<Long>
        implements ListCheckpointed<Long> {

    /**  current offset for exactly once semantics */
    private Long offset;

    /** flag for job cancellation */
    private volatile boolean isRunning = true;

    @Override
    public void run(SourceContext<Long> ctx) {
        final Object lock = ctx.getCheckpointLock();

        while (isRunning) {
            // output and state update are atomic
            synchronized (lock) {
                ctx.collect(offset);
                offset += 1;
            }
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }

    @Override
    public List<Long> snapshotState(long checkpointId, long checkpointTimestamp) {
        return Collections.singletonList(offset);
    }

    @Override
    public void restoreState(List<Long> state) {
        for (Long s : state)
            offset = s;
    }
}

• CounterSource是一個有狀態的RichParallelSourceFunction，它實現了ListCheckpointed接口，snapshotState方法返回了當前的offset，而restoreState方法則根據傳入的state來恢復本地的offset；這裏要注意，若是要在failure或者recovery的時候達到exactly-once的語義，這裏更新offset的時候要使用SourceContext.getCheckpointLock來進行同步操做

ListCheckpointed

flink-streaming-java_2.11-1.7.0-sources.jar!/org/apache/flink/streaming/api/checkpoint/ListCheckpointed.java

@PublicEvolving
public interface ListCheckpointed<T extends Serializable> {

	/**
	 * Gets the current state of the function. The state must reflect the result of all prior
	 * invocations to this function.
	 *
	 * <p>The returned list should contain one entry for redistributable unit of state. See
	 * the {@link ListCheckpointed class docs} for an illustration how list-style state
	 * redistribution works.
	 *
	 * <p>As special case, the returned list may be null or empty (if the operator has no state)
	 * or it may contain a single element (if the operator state is indivisible).
	 *
	 * @param checkpointId The ID of the checkpoint - a unique and monotonously increasing value.
	 * @param timestamp The wall clock timestamp when the checkpoint was triggered by the master.
	 *
	 * @return The operator state in a list of redistributable, atomic sub-states.
	 *         Should not return null, but empty list instead.
	 *
	 * @throws Exception Thrown if the creation of the state object failed. This causes the
	 *                   checkpoint to fail. The system may decide to fail the operation (and trigger
	 *                   recovery), or to discard this checkpoint attempt and to continue running
	 *                   and to try again with the next checkpoint attempt.
	 */
	List<T> snapshotState(long checkpointId, long timestamp) throws Exception;

	/**
	 * Restores the state of the function or operator to that of a previous checkpoint.
	 * This method is invoked when the function is executed after a failure recovery.
	 * The state list may be empty if no state is to be recovered by the particular parallel instance
	 * of the function.
	 *
	 * <p>The given state list will contain all the <i>sub states</i> that this parallel
	 * instance of the function needs to handle. Refer to the  {@link ListCheckpointed class docs}
	 * for an illustration how list-style state redistribution works.
	 *
	 * <p><b>Important:</b> When implementing this interface together with {@link RichFunction},
	 * then the {@code restoreState()} method is called before {@link RichFunction#open(Configuration)}.
	 *
	 * @param state The state to be restored as a list of atomic sub-states.
	 *
	 * @throws Exception Throwing an exception in this method causes the recovery to fail.
	 *                   The exact consequence depends on the configured failure handling strategy,
	 *                   but typically the system will re-attempt the recovery, or try recovering
	 *                   from a different checkpoint.
	 */
	void restoreState(List<T> state) throws Exception;
}

• ListCheckpointed定義了兩個接口，一個是snapshotState方法，一個是restoreState方法
• snapshotState方法，方法有個checkpointId參數，是惟一單調遞增的數字，而timestamp則是master觸發checkpoint的時間戳，該方法要返回當前的state(List結構)
• restoreState方法會在failure recovery的時候被調用，傳遞的參數爲List類型的state，方法裏頭能夠將state恢復到本地

小結

• stateful function能夠經過CheckpointedFunction接口或者ListCheckpointed接口來使用managed operator state；對於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)
• ListCheckpointed是CheckpointedFunction的限制版，它只能支持Even-split redistribution模式的list-style state
• ListCheckpointed定義了兩個方法，分別是snapshotState方法及restoreState方法；snapshotState方法在master觸發checkpoint的時候被調用，用戶須要返回當前的狀態，而restoreState方法會在failure recovery的時候被調用，傳遞的參數爲List類型的state，方法裏頭能夠將state恢復到本地

doc

• Working with State listcheckpointed
• 聊聊flink的CheckpointedFunction