食品网站建设优化案例,嘉兴网站建设wmcn,wordpress自带高亮,外贸网站品牌官网建设Checkpoint完整流程如上图所示#xff1a;
JobMaster的CheckpointCoordinator向所有SourceTask发送RPC触发一次CheckPointSourceTask向下游广播CheckpointBarrierSouceTask完成状态快照后向JobMaster发送快照结果非SouceTask在Barrier对齐后完成状态快照向JobMaster发送快照结…
Checkpoint完整流程如上图所示
JobMaster的CheckpointCoordinator向所有SourceTask发送RPC触发一次CheckPointSourceTask向下游广播CheckpointBarrierSouceTask完成状态快照后向JobMaster发送快照结果非SouceTask在Barrier对齐后完成状态快照向JobMaster发送快照结果JobMaster保存SubTask快照结果JobMaster收到所有SubTask快照结果后保存快照信息想SubTask通知Checkpoint完成
以下对整个流程具体说明。
CheckpointCoordinator
JobMaster将JobGraph转换为ExecutionGraph时如果开启Checkpoint会为ExecutionGraph生成一个CheckpointCoordinator
DefaultExecutionGraphBuilder.buildGraph//在此会将JobGraph转换为ExecutionGraphDefaultExecutionGraph::newDefaultExecutionGraph::attachJobGraph //创建ExecutionJobVertexDefaultExecutionTopology.fromExecutionGraph //创建ExecutionTopologyDefaultExecutionGraph::enableCheckpointing //创建CheckpointCoordinatorDefaultExecutionGraph::createCheckpointPlanCalculator//创建DefaultCheckpointPlanCalculatorCheckpointCoordinator::new CheckpointCoordinator封装了StateBackend和CheckpointStorage
StateBackend负责管理状态
HashMapStateBackend //内存EmbeddedRocksDBStateBackend //内存磁盘
CheckpointStorage则是负责存储StateBackend管理的状态
JobManagerCheckpointStorage //checkpoint state放入JobManager内存FileSystemCheckpointStorage //配置state.checkpoints.dir时
在为StreamTask构造SubtaskCheckpointCoordinatorImpl时会调用
CheckpointStorage::createCheckpointStorage创建CheckpointStorageAccess用于执行Checkpoint时解析状态存储位置
MemoryBackendCheckpointStorageAccess //对应JobManagerCheckpointStorageFsCheckpointStorageAccess //对应FileSystemCheckpointStorage
CheckpointCoordinator在执行状态快照时会调用
CheckpointStorageAccess::resolveCheckpointStorageLocation生成CheckpointStreamFactory用于生成读写状态数据流
MemCheckpointStreamFactory //对应JobManagerCheckpointStorageFsCheckpointStreamFactory //对应FileSystemCheckpointStorage
Checkpoint触发流程
JobMaster状态转换为running后通过CheckpointCoordinator向SourceTask发送TriggerCheckpoint
JobMaster端触发流程
JobMaster::start //RPCServer启动
JobMaster::onStart
JobMaster::startJobExecution
JobMaster::startJobMasterServices //获取RM地址后与RM建立连接
JobMaster::startScheduling
SchedulerBase::startScheduling
DefaultScheduler::startSchedulingInternal
SchedulerBase::transitionToRunningDefaultExecutionGraph::transitionToRunning //调用ExecutionGraph监听器通知状态变化CheckpointCoordinatorDeActivator::jobStatusChanges//触发checkpointCheckpointCoordinator::startCheckpointSchedulerCheckpointCoordinator::scheduleTriggerWithDelay //定时不断触发CheckpointCheckpointCoordinator::triggerCheckpointCheckpointCoordinator::startTriggeringCheckpointDefaultCheckpointPlanCalculator::calculateCheckpointPlan//Plan中会隔离出SourceTask作为作为Trigger Checkpoint的入口CheckpointCoordinator::createPendingCheckpointCheckpointCoordinator::triggerCheckpointRequestCheckpointCoordinator::triggerTasks Execution::triggerCheckpoint //向每个SourceTask发送TriggerCheckpoint请求Execution::triggerCheckpointHelperTaskManagerGateway::triggerCheckpoint//向TaskExecutor发RPCStreamTask端执行流程
SourceTask
SourceTask由JobMaster RPC直接触发执行时先广播CheckpointBarrier然后对状态执行异步快照
TaskExecutor::triggerCheckpoint
Task::triggerCheckpointBarrier
AbstractInvokable::triggerCheckpointAsync
SourceStreamTask::triggerCheckpointAsync
StreamTask::triggerCheckpointAsync
StreamTask::triggerCheckpointAsyncInMailbox
StreamTask::performCheckpoint
SubtaskCheckpointCoordinatorImpl::checkpointStateOperatorChain.broadcastEvent //广播CheckpointBarrier
CheckpointStorage::createCheckpointStorage//为JobId创建CheckpointStorageAccess
SubtaskCheckpointCoordinatorImpl::takeSnapshotSync
CheckpointStorageWorkerView::resolveCheckpointStorageLocation//CheckpointStorageAccess创建 CheckpointStreamFactoryOperatorChain::snapshotState //对每个OperatorRegularOperatorChain::buildOperatorSnapshotFuturesRegularOperatorChain::checkpointStreamOperatorAbstractStreamOperator::snapshotStateStreamOperatorStateHandler::snapshotState//调用Operator/Keyed Backend的snapshotStateSnapshotContextSynchronousImpl::newAbstractUdfStreamOperator::snapshotState //调用UDF中snapshotState方法一般用于更新OperatorStateDefaultOperatorStateBackend::snapshotSnapshotStrategyRunner::snapshotDefaultOperatorStateBackendSnapshotStrategy::syncPrepareResources//深copy operator state,便于后续进行异步快照DefaultOperatorStateBackendSnapshotStrategy::asyncSnapshot//异步快照 CheckpointStateOutputStream::closeAndGetHandleOperatorStreamStateHandle::new //包装元信息及数据StreamStateHandleHeapKeyedStateBackend::snapshotSnapshotStrategyRunner::snapshotHeapSnapshotStrategy::syncPrepareResourcesHeapSnapshotStrategy::asyncSnapshot //采用COWSateTable异步快照CheckpointStateOutputStream::closeAndGetHandleKeyGroupsStateHandle::new //包装KeyGroup及数据StreamStateHandle
SubtaskCheckpointCoordinatorImpl::finishAndReportAsync //向JobMaster发送checkpoint的结果AsyncCheckpointRunnable::new AsyncCheckpointRunnable::runAsyncCheckpointRunnable::finalizeNonFinishedSnapshotsOperatorSnapshotFinalizer::new //等待TaskSnapshot状态信息序列化完成AsyncCheckpointRunnable::reportCompletedSnapshotStatesTaskStateManagerImpl::reportTaskStateSnapshotsRpcCheckpointResponder::acknowledgeCheckpoint//向JobMaster发送Ack带上State信息非SourceTask
在StreamTask启动后调用StreamTask::processInput不断读取数据进行处理 非SourceTask在收到上游的CheckpointBarrier对齐后触发Checkpoint,
StreamTask::processInput
StreamOneInputProcessor::processInput
StreamTaskNetworkInput::emitNext(StreamTaskNetworkOutput)
AbstractStreamTaskNetworkInput::emitNext //循环不断从buffer中读取StreamElement
处理CheckpointedInputGate::pollNextCheckpointedInputGate::handleEventSingleCheckpointBarrierHandler::processBarrierSingleCheckpointBarrierHandler::markCheckpointAlignedAndTransformStateWaitingForFirstBarrier::barrierReceivedAbstractAlignedBarrierHandlerState::barrierReceivedSingleCheckpointBarrierHandler.ControllerImpl::allBarriersReceived//判断对齐AbstractAlignedBarrierHandlerState::triggerGlobalCheckpointSingleCheckpointBarrierHandler.ControllerImpl::triggerGlobalCheckpointSingleCheckpointBarrierHandler::triggerCheckpointCheckpointBarrierHandler::notifyCheckpoint //触发StreamTask CheckpointStreamTask::triggerCheckpointOnBarrierStreamTask::performCheckpoint //后续调用过程与SourceTask一样SubtaskCheckpointCoordinatorImpl::checkpointState 根据调用栈看出非SourceStreamTask执行Checkpoint只是触发时机不同SourceTask由JobMaster RPC定时不断触发非SourceTask则是在上游的CheckpointBarrier对齐后触发Checkpoint最终执行逻辑都是将当前算子的信息写入CheckpointStorage后向JobMaster发送确认信息。
StreamTask向JobMaster ACK信息中包含状态元信息及StreamStateHandle根据状态存储位置分为
ByteStreamStateHandle //对应JobManagerCheckpointStorage,将状态序列化为byte[]发送给JobMasterFileStateHandle //对应FileSystemCheckpointStorage,将状态写入文件系统后将文件路径发送给JobMaster
JobMaster端完成流程
JobMaster收到StreamTask的acknowledgeCheckpoint后
JobMaster::acknowledgeCheckpoint
SchedulerBase::acknowledgeCheckpoint
ExecutionGraphHandler::acknowledgeCheckpoint
CheckpointCoordinator::receiveAcknowledgeMessagePendingCheckpoint::acknowledgeTask //某一个Task的确认PendingCheckpoint::updateOperatorState//更新SubTask状态信息CheckpointCoordinator::completePendingCheckpoint//所有Task Ack后PendingCheckpoint::finalizeCheckpointCheckpoints.storeCheckpointMetadata//保存CheckpointMetadataCompletedCheckpoint::newCheckpointCoordinator::sendAcknowledgeMessages//向Task通知Checkpoint完成消息ExecutionVertex::notifyCheckpointCompleteTaskManagerGateway.notifyCheckpointCompleteJobMaster收到所有StreamTask的Checkpoint状态信息后标志一次Checkpoint完成这时会通知StreamTask CheckPoint完成消息便于SubTask监听Checkpoint完成后做后续动作。
