聊聊flink的CheckpointedFunction
阿新 • • 發佈:2018-12-05
序
本文主要研究一下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物件