甘肃省建设部网站,成立公司的可行性报告,如何做网站流量报告,深圳网络营销推广外包Flink checkpoint
Checkpoint是Flink实现容错机制最核心的功能#xff0c;能够根据配置周期性地基于Stream中各个Operator的状态来生成Snapshot#xff0c;从而将这些状态数据定期持久化存储下来#xff0c;从而将这些状态数据定期持久化存储下来#xff0c;当Flink程序一…Flink checkpoint
Checkpoint是Flink实现容错机制最核心的功能能够根据配置周期性地基于Stream中各个Operator的状态来生成Snapshot从而将这些状态数据定期持久化存储下来从而将这些状态数据定期持久化存储下来当Flink程序一旦意外崩溃时重新运行程序时可以有选择地从这些Snapshot进行恢复从而修正因为故障带来的程序数据状态中断。
Checkpoint指定触发生成时间间隔后每当需要触发Checkpoint时会向Flink程序运行时的多个分布式的Stream Source中插入一个Barrier标记当一个Operator接收到一个Barrier时它会暂停处理Steam中新接收到的数据记录每个Stream中都会存在对应的Barrier该Operator要等到所有的输入Stream中的Barrier都到达。当所有Stream中的Barrier都已经到达该Operator这时所有的Barrier在时间上看来是同一个时刻点表示已经对齐)该Operator会将数据记录Outgoing Records发射Emit出去作为下游Operator的输入最后将Barrier对应Snapshot发射Emit出去作为此次Checkpoint的结果数据
开启checkpoint //1.1 开启CKenv.enableCheckpointing(5000);env.getCheckpointConfig().setCheckpointTimeout(10000);env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//env.setStateBackend(new FsStateBackend(hdfs://hadoop102:8020/cdc-test/ck));
ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION表示一旦Flink处理程序被cancel后会保留Checkpoint数据以便根据实际需要恢复到指定的Checkpoint处理。
上面代码配置了执行Checkpointing的时间间隔为1分钟。
保存多个checkpoint 默认情况下如果设置了Checkpoint选项则Flink只保留最近成功生成的1个Checkpoint
Flink可以支持保留多个Checkpoint需要在Flink的配置文件conf/flink-conf.yaml中添加如下配置指定最多需要保存Checkpoint的个数
state.checkpoints.num-retained: 20
如果希望会退到某个Checkpoint点只需要指定对应的某个Checkpoint路径即可实现。
从checkpoint 恢复 如果Flink程序异常失败或者最近一段时间内数据处理错误我们可以将程序从某一个Checkpoint点比如chk-860进行回放执行如下命令
bin/flink run -s hdfs://namenode01.td.com/flink-1.5.3/flink-checkpoints/582e17d2cc343e6c56255d111bae0191/chk-860/_metadata flink-app-jobs.jar
所有的Checkpoint文件都在以Job ID为名称的目录里面当Job停掉后重新从某个Checkpoint点chk-860进行恢复时重新生成Job IDCheckpoint编号会从该次运行基于的编号继续连续生成chk-861、chk-862、chk-863
checkpoint的建议
Checkpoint 间隔不要太短 过短的间对于底层分布式文件系统而言会带来很大的压力。Flink 作业处理 record 与执行 checkpoint 存在互斥锁过于频繁的checkpoint可能会影响整体的性能。合理设置超时时间
Flink savepoint
Savepoint会在Flink Job之外存储自包含self-contained结构的Checkpoint它使用Flink的Checkpointing机制来创建一个非增量的Snapshot里面包含Streaming程序的状态并将Checkpoint的数据存储到外部存储系统中
Flink程序中包含两种状态数据:
用户定义的状态User-defined State是基于Flink的Transformation函数来创建或者修改得到的状态数据 系统状态System State是指作为Operator计算一部分的数据Buffer等状态数据比如在使用Window Function时在Window内部缓存Streaming数据记录 Flink提供了API来为程序中每个Operator设置ID这样可以在后续更新/升级程序的时候可以在Savepoint数据中基于Operator ID来与对应的状态信息进行匹配从而实现恢复。
设置Operator ID:
DataStreamString stream env.// Stateful source (e.g. Kafka) with ID.addSource(new StatefulSource()).uid(source-id) // ID for the source operator.shuffle()// Stateful mapper with ID.map(new StatefulMapper()).uid(mapper-id) // ID for the mapper// Stateless printing sink.print(); // Auto-generated ID
创建Savepoint
创建一个Savepoint需要指定对应Savepoint目录有两种方式来指定
需要配置Savepoint的默认路径需要在Flink的配置文件conf/flink-conf.yaml中添加如下配置设置Savepoint存储目录
state.savepoints.dir: hdfs://namenode01.td.com/flink/flink-savepoints
手动执行savepoint命令的时候指定Savepoint存储目录
bin/flink savepoint :jobId [:targetDirectory]
使用默认配置
bin/flink savepoint 40dcc6d2ba90f13930abce295de8d038为正在运行的Flink Job指定一个目录存储Savepoint数据
bin/flink savepoint 40dcc6d2ba90f13930abce295de8d038 hdfs://namenode01.td.com/tmp/flink/savepoints从Savepoint恢复
bin/flink run -s :savepointPath [:runArgs]以上面保存的Savepoint为例恢复Job运行
bin/flink run -s hdfs://namenode01.td.com/tmp/flink/savepoints/savepoint-40dcc6-a90008f0f82f flink-app-jobs.jar会启动一个新的Flink JobID为cdbae3af1b7441839e7c03bab0d0eefd
Savepoint 目录结构 1bbc5是Flink Job ID字符串前6个字符后面bd967f90709b是随机生成的字符串 _metadata文件包含了Savepoint的元数据信息 其他文件内容都是序列化的状态信息
总结
checkpoint和savepoint是Flink为我们提供的作业快照机制它们都包含有作业状态的持久化副本。
用几句话总结一下。 checkpoint的侧重点是“容错”即Flink作业意外失败并重启之后能够直接从早先打下的checkpoint恢复运行且不影响作业逻辑的准确性。而savepoint的侧重点是“维护”即Flink作业需要在人工干预下手动重启、升级、迁移或A/B测试时先将状态整体写入可靠存储维护完毕之后再从savepoint恢复现场。 savepoint是“通过checkpoint机制”创建的所以savepoint本质上是特殊的checkpoint。 checkpoint面向Flink Runtime本身由Flink的各个TaskManager定时触发快照并自动清理一般不需要用户干预savepoint面向用户完全根据用户的需要触发与清理。 checkpoint的频率往往比较高因为需要尽可能保证作业恢复的准确度所以checkpoint的存储格式非常轻量级但作为trade-off牺牲了一切可移植portable的东西比如不保证改变并行度和升级的兼容性。savepoint则以二进制形式存储所有状态数据和元数据执行起来比较慢而且“贵”但是能够保证portability如并行度改变或代码升级之后仍然能正常恢复。 checkpoint是支持增量的通过RocksDB特别是对于超大状态的作业而言可以降低写入成本。savepoint并不会连续自动触发所以savepoint没有必要支持增量。
