无锡网站公司哪家好wordpress酒店模板下载

无锡网站公司哪家好,wordpress酒店模板下载,wordpress字体抖动,wordpress用户前端页面前文我们梳理了 Flink 状态管理相关的源码#xff0c;我们知道#xff0c;状态是要与 Checkpoint 配合使用的。因此#xff0c;本文我们就一起来看一下 Checkpoint 相关的源码。写在前面 在Flink学习笔记#xff1a;如何做容错一文中#xff0c;我们介绍了 Flink 的 Chec…前文我们梳理了 Flink 状态管理相关的源码我们知道状态是要与 Checkpoint 配合使用的。因此本文我们就一起来看一下 Checkpoint 相关的源码。写在前面在Flink学习笔记如何做容错一文中我们介绍了 Flink 的 Checkpoint 机制。Checkpoint 分为 EXACTLY_ONCE 和 AT_LEAST_ONCE 两种模式。我们一起回顾一下一次完整的 Checkpoint 具体流程Checkpoint 是由 CheckpointCoordinator 触发Source 节点收到触发请求后会将 State 进行持久化同时向下游发送 Barrier 消息下游节点收到 Barrier 消息后也同样对 State 进行持久化和发送 Barrier 消息。当所有节点都完成持久化过程后 CheckpointCoordinator 会将一些元数据进行持久化。带着这些背景知识我们再来梳理一下 Checkpoint 相关的代码。JobManager 端触发流程JobManager 在调用DefaultExecutionGraphBuilder.buildGraph生成 ExecutionGraph 之后会调用executionGraph.enableCheckpointing方法来设置 Checkpoint 相关的配置这个方法中创建了 CheckpointCoordinator 并注册了 CheckpointCoordinatorDeActivator 这个监听它负责启动和停止 Checkpoint 的调度。当作业变成 RUNNING 状态时CheckpointCoordinator 会部署一个定时任务 ScheduledTrigger这个定时任务就是用来周期性的触发 Checkpoint。触发 Checkpoint 的核心逻辑在CheckpointCoordinator.startTriggeringCheckpoint这个方法中。这个方法中使用了多个 CompletableFuture 来完成整个流程的编排。具体流程见下图图中不同颜色代表着使用不同线程池执行。checkpointPlanFuture这是生成 Checkpoint 执行计划的 FutureCheckpoint Plan 中维护了三个关键的集合tasksToTrigger、tasksToWaitFor 和 tasksToCommitTo。tasksToTrigger 是所有的 Source 节点表示触发 Checkpoint 的节点另外两个集合都包含了全部节点分别表示等待进行 Checkpoint 的节点和等待提交的节点。pendingCheckpointCompletableFuture生成完 Checkpoint Plan 之后会创建 pendingCheckpointCompletableFuture这个 Future 中有两个执行任务分别是生成自增的 CheckpointID 和 创建 PendingCheckpoint。PendingCheckpoint 中维护了等待完成的 task 列表当所有 task 都确认完成之后PendingCheckpoint 会变成 CompletedCheckpoint。coordinatorCheckpointsComplete这个 Future 也有两个任务第一个是初始化存储路径第二个是触发所有 OperatorCoordinator Checkpoint并确认它们的状态。masterStatesComplete触发快照所有的 Master Hook这一步主要是 CheckpointCoordinator 用来收集 JobManager 级别状态。masterTriggerCompletionPromise在 masterStatesComplete 和 coordinatorCheckpointsComplete 都执行完成后会开始执行 masterTriggerCompletionPromise。masterTriggerCompletionPromise 的任务是调用 triggerCheckpointRequest 来产生 Barrier 消息。具体的触发流程见下图。至此JobManager 端的触发流程就完成了接下来就到了 TaskManager 端了。TaskManager 端执行流程进入 TaskExecutor 后具体调用过程如下图。TaskManager 的核心逻辑在SubtaskCheckpointCoordinatorImpl.checkpointState方法中。这个方法中的注释也很详细整体上分为6个步骤判断是否是需要终止的 Checkpoint如果是则向下游发送取消 Checkpoint 的广播消息。做一些前置的准备工作这一步通常情况下是一个空实现。向下游发送 Barrier 消息。注册 Alignment timer当 aligned 超时时转换为 unaligned。通知 StateWriter当前 Subtask 对输出通道的写入已经完成并提交状态句柄。异步执行状态写入并完成上报。下面我们来关注几个重点的步骤。Barrier 消息在步骤2中首先是创建 BarrierBarrier 消息包括三个部分// checkpointIdprivatefinallongid;// 时间戳privatefinallongtimestamp;// checkpoint 相关参数包括对齐类型、checkpoint 类型、目前地址privatefinalCheckpointOptionscheckpointOptions;生成 Barrier 之后会调用operatorChain.broadcastEvent进行广播消息。这里广播消息就是向下游所有的节点的所有 ResultSubpartition 发送。状态写入SubtaskCheckpointCoordinatorImpl.takeSnapshotSync方法用来构建 OperatorSnapshotFutures 中的四个 Future每个 Future 的任务是为不同类型的 State 提供写入逻辑。NonnullprivateRunnableFutureSnapshotResultKeyedStateHandlekeyedStateManagedFuture;NonnullprivateRunnableFutureSnapshotResultKeyedStateHandlekeyedStateRawFuture;NonnullprivateRunnableFutureSnapshotResultOperatorStateHandleoperatorStateManagedFuture;NonnullprivateRunnableFutureSnapshotResultOperatorStateHandleoperatorStateRawFuture;在底层逻辑中会为每个 Operator 设置对应的 State 的 Future。具体调用流程如下设置好这些 Future 之后会在finishAndReportAsync方法中创建 AsyncCheckpointRunnable 线程调用 get 来获取执行结果拿到执行结果后会将 Checkpoint 信息上报给 CheckpointCoordinator。JobManager 端确认流程TaskManager 通过调用checkpointCoordinatorGateway.acknowledgeCheckpoint上报 Checkpoint 信息后流程就又回到 JobManager 了。JobManager 的确认流程主要做了两件事将 pendingCheckpoint 转换成 completedCheckpoint在这个转换过程中还做了清理过期 Checkpoint 和持久化元数据等操作。向所有 commit 的 Task 发送 Checkpoint 完成的通知。收到这个通知后大部分 Task 没有什么特殊逻辑也有一部分 Source 或者 Sink 会做提交事务等操作。至此JobManager 和 Source 端算子的一次 Checkpoint 就完成了。接下来我们再看一下非 Source 节点是如何做 Checkpoint 的。非 Source 节点处理流程非 Source 节点处理 Barrier 的入口和处理业务数据的入口相同都是StreamTask.processInput方法。我们还是先来看具体的调用流程。跟着调用链路我们一路找到了 processBarrier 方法这里区分了两个 barrierHandler。SingleCheckpointBarrierHandler 负责处理 EXACTLY_ONCE 语义CheckpointBarrierTracker 负责处理 AT_LEAST_ONCE 语义。EXACTLY_ONCEEXACTLY_ONCE 在处理 Barrier 的逻辑如下如果只有一个 channel就立即触发 Checkpoint。如果有多个 channel分为三种情况a) 如果收到的是第一个 channel标记开始进行 barrier 对齐并阻塞 channel。b) 如果不是第一个 channel也不是最后一个 channel只对 channel 进行阻塞。c) 如果收到最后一个 channel就会触发 Checkpoint并取消所有 channel 阻塞状态。这里触发的逻辑与 Source 节点相同通过调用链路可以一直找到 performCheckpoint。AT_LEAST_ONCEAT_LEAST_ONCE 处理 Barrier 的逻辑如下如果只有一个 channel就立即触发 Checkpoint。如果有多个 channel同样分为三种情况a) 如果收到的是第一个 channel则更新当前 checkpointID标记开始 barrier 对齐。b) 如果收到的不是第一个 channel也不是最后一个 channel就只做计数。c) 如果收到的是最后一个 channel就会开始触发 Checkpoint。这里触发逻辑也是调用 performCheckpoint与 Source 节点逻辑相同。总结本文我们梳理了 Checkpoint 的源码逻辑。最开始由 JobManager 中的 CheckpointCoordinator 进行调度并向 TaskManager 发送触发请求。Source 节点收到请求后会向下游发送 Barrier 消息然后写入状态数据和上报 Checkpoint 信息。CheckpointCoordinator 收集完确认消息后会持久化元数据并通知所有 Task 完成 commit。最后还分别介绍了 EXACTLY_ONCE 和 AT_LEAST_ONCE 模式下非 Source 节点的处理逻辑。这里埋一个 Hook状态数据写入逻辑的细节我们没有深入了解会在下篇进行深入分析。