flink1.7 checkpoint源码分析
初始化state类
//org.apache.flink.streaming.runtime.tasks.StreamTask#initializeState
initializeState();
private void initializeState() throws Exception {
?
StreamOperator[] allOperators = operatorChain.getAllOperators();
?
for (StreamOperator operator : allOperators) {
if (null != operator) {
operator.initializeState();
}
}
}
operator.initializeState() 调用的方法路径 org.apache.flink.streaming.api.operators.AbstractStreamOperator#initializeState() ，所有的操作流类都继承该类，同时也没有重写这个方法。
public final void initializeState() throws Exception {
////这里会调用状态后端，里面很重要
1. final StreamOperatorStateContext context =
streamTaskStateManager.streamOperatorStateContext(
getOperatorID(),
getClass().getSimpleName(),
this,
keySerializer,
streamTaskCloseableRegistry,
metrics);
...
|
streamTaskStateManager.streamOperatorStateContext(......)调用方法的路径org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl#streamOperatorStateContext
......
// -------------- Keyed State Backend 这里是重点 关于checkpoint--------------
keyedStatedBackend = keyedStatedBackend(
keySerializer,
operatorIdentifierText,
prioritizedOperatorSubtaskStates,
streamTaskCloseableRegistry,
metricGroup);
?
// -------------- Operator State Backend 这里是重点 关于checkpoint --------------
operatorStateBackend = operatorStateBackend(
operatorIdentifierText,
prioritizedOperatorSubtaskStates,
streamTaskCloseableRegistry);
......
keyedStatedBackend() 这个方法最里面是调用了 org.apache.flink.streaming.api.operators.BackendRestorerProcedure#attemptCreateAndRestore
private T attemptCreateAndRestore(Collection restoreState) throws Exception {
......
// create a new, empty backend.
final T backendInstance = instanceSupplier.get();
?
// attempt to restore from snapshot (or null if no state was checkpointed).
backendInstance.restore(restoreState);
......
}
backendInstance.restore(restoreState)调用的方法路径org.apache.flink.runtime.state.DefaultOperatorStateBackend#restore
// registeredOperatorStates这个对象是核心
...
PartitionableListState<?> listState = registeredOperatorStates.get(restoredSnapshot.getName());
?
if (null == listState) {
listState = new PartitionableListState<>(restoredMetaInfo);
//重点，这里就是存储了快照状态类
//********************************************************************
registeredOperatorStates.put(listState.getStateMetaInfo().getName(), listState);
//********************************************************************
} else {
// TODO with eager state registration in place, check here for serializer migration strategies
}
...
triggerCheckpoint 将定时触发执行checkpoint，而上面是是初始化的执行逻辑
定时快照state类
org.apache.flink.runtime.checkpoint.CheckpointCoordinator#triggerCheckpoint(long, boolean)
......
// send the messages to the tasks that trigger their checkpoint 我猜测这里就是远程发送触发checkpoint的步骤 这里进行的数据文件的生成奶奶的
for (Execution execution: executions) {
execution.triggerCheckpoint(checkpointID, timestamp, checkpointOptions);
}
......
execution.triggerCheckpoint调用路径 org.apache.flink.runtime.executiongraph.Execution#triggerCheckpoint
/**
* Trigger a new checkpoint on the task of this execution.
* @param checkpointId of th checkpoint to trigger
* @param timestamp of the checkpoint to trigger
* @param checkpointOptions of the checkpoint to trigger
/
public void triggerCheckpoint(long checkpointId, long timestamp, CheckpointOptions checkpointOptions) {
......
final LogicalSlot slot = assignedResource;
if (slot != null) {
final TaskManagerGateway taskManagerGateway = slot.getTaskManagerGateway();
taskManagerGateway.triggerCheckpoint(attemptId, getVertex().getJobId(), checkpointId, timestamp, checkpointOptions);
}
.....
}
taskManagerGateway.triggerCheckpoint(......)里面最终调用路径 org.apache.flink.runtime.taskexecutor.TaskExecutor#triggerCheckpoint
@Override
public CompletableFuture//entry.getValue() 里面就是状态类 将状态类存储在新建的map对象中
if (!registeredOperatorStates.isEmpty()) {
for (Map.Entry
if (null != listState) {
listState = listState.deepCopy();
}
registeredOperatorStatesDeepCopies.put(entry.getKey(), listState);
}
}
//广播状态
if (!registeredBroadcastStates.isEmpty()) {
for (Map.Entry
BackendWritableBroadcastState broadcastState = entry.getValue();
if (null != broadcastState) {
broadcastState = broadcastState.deepCopy();
}
registeredBroadcastStatesDeepCopies.put(entry.getKey(), broadcastState);
}
}
}
//这个方法里面生成了状态数据文件
AsyncSnapshotCallable> snapshotCallable =
new AsyncSnapshotCallable>() {
?
@Override
protected SnapshotResult
......
// get the registered operator state infos ...
List
new ArrayList<>(registeredOperatorStatesDeepCopies.size());
?
for (Map.Entry
registeredBroadcastStatesDeepCopies.entrySet()) {
broadcastMetaInfoSnapshots.add(entry.getValue().getStateMetaInfo().snapshot());
}
?
// ... write them all in the checkpoint stream ...
DataOutputView dov = new DataOutputViewStreamWrapper(localOut);
?
OperatorBackendSerializationProxy backendSerializationProxy =
new OperatorBackendSerializationProxy(operatorMetaInfoSnapshots, broadcastMetaInfoSnapshots);
?
backendSerializationProxy.write(dov);
?
// ... and then go for the states ...
?
......
}
};
?
final FutureTask
snapshotCallable.toAsyncSnapshotFutureTask(closeStreamOnCancelRegistry);
?
if (!asynchronousSnapshots) {
task.run();
}
?
return task;
}
}
从上面我们可以看到，状态类都存放在registeredOperatorStatesDeepCopies这个map中。
用户能够更新状态类的数据都是因为这样访问到了状态类
public void initializeState(FunctionInitializationContext context) throws Exception {
......
checkpointedState = context.getOperatorStateStore().getListState(descriptor);
......
}
调用的就是org.apache.flink.runtime.state.DefaultOperatorStateBackend#getListState(org.apache.flink.api.common.state.ListStateDescriptor)
/**
* @Description: 返回状态类的时候，将状态类放入map对象供后面写入文件中
* @Param:
* @return:
* @Author: intsmaze
* @Date: 2019/1/18
/
private ListState getListState(
ListStateDescriptor stateDescriptor,
OperatorStateHandle.Mode mode) throws StateMigrationException {
@SuppressWarnings("unchecked")
PartitionableListState previous = (PartitionableListState) accessedStatesByName.get(name);
if (previous != null) {
checkStateNameAndMode(
previous.getStateMetaInfo().getName(),
name,
previous.getStateMetaInfo().getAssignmentMode(),
mode);
return previous;
}
......
PartitionableListState partitionableListState = (PartitionableListState) registeredOperatorStates.get(name);
?
if (null == partitionableListState) {
// no restored state for the state name; simply create new state holder
partitionableListState = new PartitionableListState<>(
new RegisteredOperatorStateBackendMetaInfo<>(
name,
partitionStateSerializer,
mode));
//这里也会存储状态类数据registeredOperatorStates这个对象和DefaultOperatorStateBackendSnapshotStrategy类的快照方法访问的对象共享
//************************************************************
registeredOperatorStates.put(name, partitionableListState);
}
Powered by .NET 6 on Kubernetes