网站建设课程的感想,网站建设与管理适合女生学吗,健康呼伦贝尔二维码高清,加强网站政务服务建设检查点与保存点 检查点Checkpoint概述保存时机保存与恢复检查点算法 检查点配置启用检查点指定存储位置其它配置通用增量 保存点Savepoint概述使用保存点切换状态后端 SQL客户端中操作提交作业触发恢复 检查点Checkpoint
概述 在 Flink 中#xff0c;检查点是用于实现状态一致… 检查点与保存点 检查点Checkpoint概述保存时机保存与恢复检查点算法 检查点配置启用检查点指定存储位置其它配置通用增量 保存点Savepoint概述使用保存点切换状态后端 SQL客户端中操作提交作业触发恢复 检查点Checkpoint
概述 在 Flink 中检查点是用于实现状态一致性和故障恢复的关键机制。检查点功能可确保作业的状态在发生故障时能够进行可靠地恢复。 检查点具有以下特性
状态一致性检查点会将作业的状态数据保存在持久化存储中以确保数据的一致性。通过在特定时间点对作业状态进行快照可以捕获整个作业的状态容错性当作业发生故障时Flink可以使用最近的检查点来恢复作业的状态。恢复作业时Flink会从最近一个成功完成的检查点开始进行恢复操作同步机制为了确保一致性在触发检查点时Flink会暂停作业中的所有计算任务并将状态快照写入持久化存储中。一旦检查点完成作业的计算任务将会继续执行检查点模式Flink提供两种检查点模式Exactly-once确保数据在一次计算中只会被处理一次并且状态可以准确地恢复到检查点的状态At-least-once允许数据在一次计算中可能被处理多次但可以在发生故障后恢复到检查点的状态高可用性Flink 支持将检查点数据存储在分布式文件系统如 HDFS上以提供高可用性和容错能力保存时机 在Flink中检查点的保存是通过配置检查点间隔和触发条件来控制 检查点间隔 Checkpoint Interval
可以配置一个时间间隔在该间隔时间过后Flink 将自动触发一个新的检查点检查点间隔可以根据应用程序的需求进行调整通常是根据数据处理的延迟和吞吐量来决定较短的检查点间隔意味着更频繁的检查点但可能会增加系统开销和延迟而较长的间隔则会减少开销但会增加数据丢失的潜在风险触发条件 Triggering Condition
精确一次处理 Exactly-Once Processing在 Flink 中可以通过配置“精确一次处理”的语义来触发检查点。这意味着只有在前一个检查点成功完成后才会触发新的检查点确保数据的一致性和完整性。基于数据量 Data Size可以设置一个数据大小阈值在接收到一定数量的记录后触发检查点。这个阈值可以根据应用程序的需求进行调整以平衡开销和一致性。基于时间 Time Based可以设置一个时间阈值在一定时间间隔内没有触发检查点时强制触发一个检查点。这用于确保即使在没有活动数据的情况下也能定期保存状态数据。保存与恢复
Flink 中检查点的保存过程大致分为以下几个步骤
1.触发检查点Flink会根据配置的触发条件例如时间间隔或数据量自动触发一个新的检查点。或者可以通过编程方式在代码中显式触发检查点。2.暂停计算任务在触发检查点时Flink会暂停应用程序的所有计算任务确保没有新的状态更新。3.状态快照一旦计算任务暂停Flink将应用程序的状态数据进行快照snapshot包括Keyed State和Operator State。这些状态数据会被写入到持久化存储中。4.存储检查点数据Flink将状态快照持久化存储到配置的检查点存储后端中。具体存储的方式取决于所选择的存储后端例如写入到分布式文件系统、对象存储服务或远程文件系统。5.恢复计算任务一旦状态数据持久化完成Flink 将恢复应用程序的计算任务继续处理输入数据。在发生故障时Flink可以使用已保存的检查点数据进行应用程序的恢复过程如下
1.检测到故障Flink监控作业状态时会检测到故障例如计算节点崩溃。2.选择最近的检查点一旦故障被检测到Flink会选择最近成功完成的检查点。3.从检查点恢复状态Flink会从所选检查点的存储位置中读取状态数据并将其加载到应用程序中恢复应用程序的状态。4.恢复计算任务一旦状态数据加载完成Flink 将恢复应用程序的计算任务从故障发生时的状态继续处理输入数据。检查点算法 在Flink中采用了基于Chandy-Lamport算法的分布式快照可以在不暂停整体流处理的前提下将状态备份保存到检查点。 Flink的检查点算法是基于快照的机制实现的它是一种增量快照算法称为异步屏障asynchronous barrier算法用于实现检查点的一致性。 通过异步屏障算法Flink可以实现在分布式环境中对应用程序状态的一致性保存确保在故障发生时能够正确地恢复状态并继续进行数据处理。这种算法的设计考虑了分布式环境下的并发和异步处理的特点以保证高效且可靠的检查点机制。 检查点算法的基本流程
触发检查点根据配置的触发条件Flink触发一个新的检查点异步屏障插入在所有算子之间插入屏障barrier操作。屏障是一种特殊的记录在流中传播并且在到达算子后会阻塞输入直到在所有流并行分区的所有算子上都收到屏障为止等待所有屏障一旦插入屏障所有算子将等待所有输入流上的屏障。这可以确保在屏障之后的所有记录都是一致的并且有效防止任何后续状态更新状态快照在等待所有屏障后Flink将进行实际的状态快照操作。它会在算子中将状态数据写入到持久化存储中以进行检查点保存完成屏障传播在状态快照完成后屏障将被传播到下游算子通知它们可以继续处理记录恢复计算任务在发生故障时Flink使用已保存的检查点数据进行应用程序的恢复通过加载状态数据来恢复应用程序的状态检查点分界线 检查点分界线Barrier是在Flink中用于保证检查点的一致性的特殊记录。在Flink的异步屏障算法中屏障插入和分界线的概念是相关的。 当触发一个新的检查点时Flink 在流中的所有算子之间插入屏障操作。屏障是一种特殊的记录会在流中传播并且在到达算子后会阻塞输入直到在所有流并行分区的所有算子上都收到屏障为止。 屏障的到达具有两个重要的作用
一致性分界线屏障的到达表示在该位置之前的所有记录都被视为属于之前的检查点而之后的记录将属于新检查点。这样屏障将流分割为多个一致性分段确保在一个检查点完成后新的检查点将在屏障之后开始。阻塞输入屏障的到达会阻塞输入直到所有输入流并行分区上的算子都收到屏障为止。这确保了在算子处理记录之前所有输入数据都已经达到一致的检查点位置。通过插入屏障和处理屏障的到达Flink 可以保证检查点的一致性和状态快照的完整性。屏障作为检查点的分界线将流划分为单个一致性分段并确保所有算子在检查点完成后开始处理新的数据记录。 Barrier对齐的精准一次 分布式快照算法之精确一次Exactly-once语义它结合了屏障对齐Barrier Alignment和状态快照State Snapshotting来实现精确一次的语义保证。 过程如下
屏障对齐在启动检查点时Flink 将在流的所有并行分区之间插入屏障barrier。屏障的到达通知所有算子在其上的输入都已经到达一致的检查点位置。状态快照在屏障对齐后Flink 使用状态快照机制将算子的状态数据保存到持久化存储中形成一个一致的检查点。检查点确认在状态快照完成后Flink 会等待所有参与分布式快照的任务确认其已完成检查点。这可以确保所有任务已经成功保存了状态数据。任务协调器通知一旦所有任务都确认完成检查点任务协调器将发送一个通知表示检查点已经完成。恢复当发生故障时Flink 可以使用已保存的检查点数据进行应用程序的恢复。它将加载检查点中的状态数据并通过屏障对齐来确保在故障点之后的记录不会被处理。该算法有效地保证了在故障发生时可以正确地恢复应用程序的状态并避免了重复处理数据或丢失数据的情况从而实现了精确一次的语义保证。 一个Task收到所有上游同一个编号的barrier之后才会对自己的本地状态做备份在barrier对齐过程中barrier后面的数据阻塞等待,不会越过barrier Barrier对齐的至少一次 分布式快照算法之至少一次At-least-once语义它是通过基于异步屏障Asynchronous Barrier的检查点算法来实现的。 过程如下
触发检查点根据配置的触发条件Flink 触发一个新的检查点。异步屏障插入在所有算子之间插入异步屏障barrier操作。屏障是特殊记录在流中传播并且在到达算子后会阻塞输入直到在所有流并行分区的所有算子上都收到屏障为止。检查点确认在屏障插入之后Flink 会等待所有参与检查点的任务确认它们已经处理了该屏障。一旦收到所有任务的确认就可以继续进行下一步。状态快照在检查点确认之后Flink 执行实际的状态快照操作。它会在算子中将状态数据写入持久化存储以进行检查点保存。完成屏障传播在状态快照完成后屏障将被传播到下游算子通知它们可以继续处理记录。该算法尽管在故障发生时可能会出现一些重复处理的情况但它确保了至少一次的语义即每个记录在最多一次被处理。这是一种容错机制可以保证在发生故障时能够正确地恢复应用程序的状态确保数据被至少一次地处理。 一个Task收到所有上游同一个编号的barrier之后才会对自己的本地状态做备份在barrier对齐过程中先到的barrier其后面的数据不阻塞接着计算 非Barrier对齐的精准一次 分布式快照算法之非Barrier对齐的精确一次语义它使用了异步处理与检查点保存的方式来实现精确一次的语义。 过程如下
触发检查点根据配置的触发条件Flink 触发一个新的检查点。非Barrier对齐不像Barrier对齐算法那样在流中插入屏障非Barrier对齐算法利用异步处理和状态保存来实现。在每个算子中Flink 会使用异步的方式执行状态快照并将状态数据保存到持久化存储中形成一个一致的检查点。检查点确认一旦算子完成了状态快照Flink 会等待所有参与检查点的任务确认它们已经完成了状态保存。这可以确保所有任务都成功保存了状态数据。任务协调器通知当所有任务确认已完成检查点后任务协调器将发送一个通知表示检查点已经完成。恢复当发生故障时Flink 可以使用已保存的检查点数据进行应用程序的恢复。它会从持久化存储中加载检查点中的状态数据并通过异步处理来确保在故障点之后的记录不会被重复处理。该算法利用异步处理和检查点保存来实现状态一致性并通过恢复机制来处理故障情况从而确保应用程序的状态能够正确地恢复。 一个Task收到第一个barrier时就开始执行备份能保证精准一次
先到的barrier将本地状态备份其后面的数据接着计算输出未到的barrier其前面的数据接着计算输出同时也保存到备份中最后一个barrier到达该Task时这个Task的备份结束检查点配置 检查点的作用是为了故障恢复不能因为保存检查点占据了大量时间、导致数据处理性能明显降低。为了兼顾容错性和处理性能可以在代码中对检查点进行各种配置。 启用检查点 Flink程序是默认禁用检查点的为Flink应用开启自动保存快照的功能需要在代码中显式地调用执行环境的enableCheckpointing()方法 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();// 状态检查点间隔单位为毫秒表示周期性保存检查点的间隔时间。默认间隔周期500毫秒已经被弃用。// 每隔1秒启动一次检查点保存env.enableCheckpointing(1000);// 启用检查点: 默认是barrier对齐的周期为5s, 精准一次env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);// 周期为5s, 至少一次env.enableCheckpointing(5000, CheckpointingMode.AT_LEAST_ONCE);检查点的间隔时间是对处理性能和故障恢复速度的一个权衡。
如果希望对性能的影响更小可以调大间隔时间如果希望故障重启后迅速赶上实时的数据处理就需要将间隔时间设小一些指定存储位置 检查点具体的持久化存储位置取决于检查点存储的设置。默认情况下检查点存储在JobManager的堆内存中。而对于大状态的持久化保存Flink也提供了在其他存储位置进行保存的接口。 Flink持久化存储位置主要提供了两种作业管理器的堆内存和文件系统。通过调用检查点配置的setCheckpointStorage()来配置传入一个CheckpointStorage的实现类。
1.配置存储检查点到JobManager堆内存
StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();
CheckpointConfig checkpointConfig env.getCheckpointConfig();
checkpointConfig.setCheckpointStorage(new JobManagerCheckpointStorage());2.配置存储检查点到文件系统 实际生产应用一般会将CheckpointStorage配置为高可用的分布式文件系统
StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();
CheckpointConfig checkpointConfig env.getCheckpointConfig();
checkpointConfig .setCheckpointStorage(new FileSystemCheckpointStorage(hdfs://node01:9000/flink/checkpoints));其它配置 检查点还有很多可以配置的选项可以通过获取检查点配置来进行设置。 CheckpointConfig checkpointConfig env.getCheckpointConfig();// 超时时间: 默认10分钟。用于指定检查点保存的超时时间超时没完成就会被丢弃掉。传入一个长整型毫秒数作为参数表示超时时间checkpointConfig.setCheckpointTimeout(60000);// 最大并发检查点数量用于指定运行中的检查点最多可以有多少个// 由于每个任务的处理进度不同完全可能出现后面的任务还没完成前一个检查点的保存、前面任务已经开始保存下一个检查点。这个参数就是限制同时进行的最大数量checkpointConfig.setMaxConcurrentCheckpoints(1);// 最小间隔时间: 用于指定在上一个检查点完成之后检查点协调器最快等多久可以出发保存下一个检查点的指令。当指定这个参数时实际并发为1// 直白说就是上一轮checkpoint结束 到 下一轮checkpoint开始之间的间隔checkpointConfig.setMinPauseBetweenCheckpoints(1000);// 开启外部持久化存储: 用于开启检查点的外部持久化而且默认在作业失败的时候不会自动清理// 传入的参数ExternalizedCheckpointCleanup指定了当作业取消的时候外部的检查点该如何清理,即数据是否保留在外部系统// DELETE_ON_CANCELLATION: 在作业取消的时候会自动删除外部检查点但是如果是作业失败退出则会保留检查点// RETAIN_ON_CANCELLATION: 作业取消的时候也会保留外部检查点checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);// 检查点连续失败次数:用于指定检查点连续失败的次数当达到这个次数作业就失败退出。默认为0checkpointConfig.setTolerableCheckpointFailureNumber(10);// 开启非对齐检查点: 不再执行检查点的分界线对齐操作启用之后可以大大减少产生背压时的检查点保存时间// 设置要求检查点模式必须为精准一次exctly-once并且最大并发的检查点个数为1checkpointConfig.enableUnalignedCheckpoints();// 对齐检查点超时时间: 参数只有在启用非对齐检查点的时候有效// 默认是0表示一开始就直接用非对齐检查点。如果设置大于0一开始会使用对齐的检查点// 当对齐时间超过该参数设定的时间则会自动切换成非对齐检查点checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(1));通用增量 Flink的增量检查点是一种优化技术用于减少检查点所需的存储和处理成本。传统的全量检查点需要在每个检查点时保存整个应用程序的状态。而增量检查点仅保存自上一个检查点以来更改的状态部分从而减少了存储和处理的负担。 通用增量检查点是 Flink1.11版本引入的一项功能。它通过结全量检查点Full Checkpoint和增量检查点Incremental Checkpoint类型的检查点来实现增量检查点 具体的增量检查点流程如下
全量检查点在首次触发检查点时会执行全量检查点将整个应用程序的状态保存为检查点。增量检查点自上一个检查点以来当应用程序状态发生更改时Flink 会执行增量检查点仅保存这些更改的状态部分。这些更改的状态会与上一个全量检查点的状态进行比较只保留与之不同的部分。检查点结果最终的检查点结果包含上一个全量检查点的状态和增量检查点中的更改部分。对于故障恢复 Flink 可以使用这些检查点结果进行恢复操作。首先将上一个全量检查点的状态加载到内存中然后依次应用增量检查点的更改部分即可恢复应用程序的状态。 优缺点 增量检查点可以减少存储和传输的数据量从而降低了检查点产生和恢复的开销。然而由于增量检查点需要进行状态更改的计算和处理所以会引入一些额外的计算开销 注意Flink1.15之前只有RocksDB支持增量快照开启方式
EmbeddedRocksDBStateBackend backend new EmbeddedRocksDBStateBackend(true);方式一配置文件指定
# 启用增量检查点
state.backend.changelog.enabled: true
state.backend.changelog.storage: filesystem# 存储changelog数据
dstl.dfs.base-path: hdfs://node01:9000/changelog
execution.checkpointing.max-concurrent-checkpoints: 1
execution.savepoint-restore-mode: CLAIM方式二在代码中设置 需要引入依赖
dependencygroupIdorg.apache.flink/groupIdartifactIdflink-statebackend-changelog/artifactIdversion${flink.version}/versionscoperuntime/scope
/dependency开启changelog
env.enableChangelogStateBackend(true);保存点Savepoint
概述 在Flink中保存点Savepoints是一种用于长期存储应用程序状态的机制。保存点可以用于在应用程序遇到故障或需要更改应用程序代码时执行重启和升级。 保存点是一个检查点的特殊形式它可以用于在不同的 Flink 版本之间迁移作业状态。保存点是跨版本的状态快照允许将作业从一个版本的 Flink 迁移到另一个版本而不丢失状态。保存点通常用于进行作业升级、实验和版本回退。 保存点与检查点区别
保存点与检查点最大的区别就是触发的时机
检查点是由Flink自动管理的定期创建发生故障之后自动读取进行恢复保存点不会自动创建必须由用户明确地手动触发保存操作。保存点的用途 保存点可以当作一个强大的运维工具来使用。可以在需要的时候创建一个保存点然后停止应用做一些处理调整之后再从保存点重启。 应用场景
更新应用程序作业升级调整并行度实验和测试版本回退暂停应用程序故障排查和分析算子ID 算子ID可以在代码中直接调用。对于没有设置ID的算子Flink默认会自动进行设置在重新启动应用后可能会导致ID不同而无法兼容以前的状态。 在程序中为每一个算子手动指定ID
DataStreamString stream env.addSource(new StatefulSource()).uid(source-id).map(new StatefulMapper()).uid(mapper-id).print();使用保存点
1.创建保存点
在命令行中为运行的作业创建一个保存点镜像
jobId需要填充要做镜像保存的作业IDtargetDirectory目标路径可选表示保存点存储的路径示例如下
bin/flink savepoint :jobId [:targetDirectory]对运行的作业创建保存点在停掉一个作业时直接创建保存点
bin/flink stop --savepointPath [:targetDirectory] :jobId2.保存点路径
1.可以通过配置文件flink-conf.yaml来修改保存点的默认路径
state.savepoints.dir: hdfs://node01:9000/flink/savepoints2.对于单独的作业可以在程序代码中通过执行环境来设置
env.setDefaultSavepointDir(hdfs://node01:9000/flink/savepoints);3.从保存点重启应用
-s参数指定保存点的路径runArgs其它启动时的参数bin/flink run -s :savepointPath [:runArgs]切换状态后端 使用savepoint恢复状态的时候可以更换状态后端。建议不要在代码中指定状态后端 通过配置文件来配置或者-D参数配置。 1.提交flink作业
bin/flink run-application -d -t yarn-application -Dstate.backendhashmap -c cn.ybzy.demo.SavepointDemo FlinkDemo.jar2.停止flink作业时触发保存点
方式一stop优雅停止并触发保存点要求source实现StoppableFunction接口
bin/flink stop -p savepoint路径 job-id -yid application-id3.从savepoint恢复作业同时修改状态后端
bin/flink run-application -d -t yarn-application -s hdfs://node01:9000/flink/savepoint-26c8e0-9e0b6cd976a4 -Dstate.backendrocksdb -c cn.ybzy.demo.SavepointDemo FlinkDemo.jar 4.从保存下来的checkpoint恢复作业
bin/flink run-application -d -t yarn-application -Dstate.backendrocksdb -s hdfs://node01:9000/flink/4d8435f6be2a1d9e0b6cd976a24f6c8e/chk-175 -c cn.ybzy.demo.SavepointDemo ./FlinkDemo.jarSQL客户端中操作
提交作业
提交一个insert作业可以给作业设置名称
INSERT INTO tb_test select * from datagen;查看job列表
SHOW JOBS;触发
设置检查点、保存点路径
SET state.checkpoints.dirhdfs://node01:9000/ck;
SET state.savepoints.dirhdfs://node01:9000/sp;停止作业触发savepoint
STOP JOB 4d8435f6be2a1d9e0b6cd976a24f6c8e WITH SAVEPOINT;恢复
设置恢复的路径然后提交sql就会从savepoint恢复
SET execution.savepoint.pathhdfs://node01:9000/sp/savepoint-26c8e0-9e0b6cd976a4; --允许跳过无法还原的保存点状态
set execution.savepoint.ignore-unclaimed-state true; 跳过无法还原的保存点状态
set execution.savepoint.ignore-unclaimed-state true; 使用RESET命令重置execution.savepoint.path配置其指定后会影响后面执行的所有DML语句
RESET execution.savepoint.path;
