赞
踩
flink为了保证程序操作状态的容错性,flink在内部引入了checkpoint机制,检查点最终从datastream中以exactly once或者at least one进行状态映射,容错机制,持续的以一定时间间隔描绘整个分布式流的状态并保存在:MasterNode的内存或者HDFS文件系统,这是非常轻量级的,并不会对性能造成影响。
一旦程序跑失败后(由于机器、网络、软件等原因),Flink会停止所有的分布式streaming dataflow,系统重启所有的算子并且重新设置他们保存成功的最近检查点,input streams被设置到之前的快照状态,然后任何本应该被并行处理的数据流自动将不会成为checkpoint的一部分。
在默认状态下checkpoint机制是不开启的,我们可以通过以下方式进行开启:
env.enableCheckpoint(60000)// 60000代表checkpoint的时间间隔
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE) //设置语义
Flink’s mechanism for drawing these snapshots is described in “Lightweight Asynchronous Snapshots for Distributed Dataflows”. It is inspired by the standard Chandy-Lamport algorithm for distributed snapshots and is specifically tailored to Flink’s execution model。
checkpoint在不同的阶段是相互隔离,但是从stream从上往下有关联的传递,这里就涉及到barrier的概念
不同算子的barrier可以并行处理。
barriers作为data stream的一部分,barrier不会赶超真是数据,一般都是跟随在真实数据之后,每个barrier都带有各自的id作为checkpoint的阶段标志,一个barrier代表当前stream的checkpoint结束。
同时barries在不同的算子阶段是从上往下进行传递的,如下图:
所有的快照存储是通过stateBackEnd决定的(memory、fs、rocksdb)
env.setStateBackend() //用来设置backend的策略
/** * 指定checkpoint重启策略 * 1。Fixed Delay Restart Strategy * restart-strategy.fixed-delay.attempts: 3 * restart-strategy.fixed-delay.delay: 10 s * * 2。Failure Rate Restart Strategy * restart-strategy.failure-rate.max-failures-per-interval: 3 * restart-strategy.failure-rate.failure-rate-interval: 5 min * restart-strategy.failure-rate.delay: 10 s * * 3.No Restart Strategy *restart-strategy: none * * 4。Fallback Restart Strategy(自动选择重启策略,默认为:Fixed Delay Restart Strategy, 非常有用) * The cluster defined restart strategy is used. This helpful for streaming program which enable checkpointing. * Per default, a fixed delay restart strategy is chosen if there is no other restart strategy defined. */ env.setRestartStrategy(RestartStrategies.fixedDelayRestart( 3, // number of restart attempts Time.of(10, TimeUnit.SECONDS) // delay )) env.setRestartStrategy(RestartStrategies.failureRateRestart( 3, // 允许失败3次 Time.of(5, TimeUnit.MINUTES), // 5分钟内允许失败3次 Time.of(10, TimeUnit.SECONDS) // 没10s尝试重启一次 ))
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。