| 注册
请输入搜索内容

热门搜索

Java Linux MySQL PHP JavaScript Hibernate jQuery Nginx
mthj8765
7年前发布

Apache Flink 容错机制

   <h2>Introduce</h2>    <p>Apache Flink 提供了可以恢复数据流应用到一致状态的容错机制。确保在发生故障时,程序的每条记录只会作用于状态一次(exactly-once),当然也可以降级为至少一次(at-least-once)。</p>    <p>容错机制通过持续创建分布式数据流的快照来实现。对于状态占用空间小的流应用,这些快照非常轻量,可以高频率创建而对性能影响很小。流计算应用的状态保存在一个可配置的环境,如:master 节点或者 HDFS上。</p>    <p>在遇到程序故障时(如机器、网络、软件等故障),Flink 停止分布式数据流。系统重启所有 operator ,重置其到最近成功的 checkpoint。输入重置到相应的状态快照位置。保证被重启的并行数据流中处理的任何一个 record 都不是 checkpoint 状态之前的一部分。</p>    <p><em>注意:</em> 为了容错机制生效,数据源(例如 queue 或者 broker)需要能重放数据流。 <a href="/misc/goto?guid=4958531873897074375" rel="nofollow,noindex">Apache Kafka</a> 有这个特性,Flink 中 Kafka 的 connector 利用了这个功能。</p>    <p><em>注意:</em> 由于 Flink 的 checkpoint 是通过分布式快照实现的,接下来我们将 snapshot 和 checkpoint 这两个词交替使用。</p>    <h2>Checkpointing</h2>    <p>Flink 容错机制的核心就是持续创建分布式数据流及其状态的一致快照。这些快照在系统遇到故障时,充当可以回退的一致性检查点(checkpoint)。 Lightweight Asynchronous Snapshots for Distributed Dataflows 描述了Flink 创建快照的机制。此论文是受分布式快照算法 Chandy-Lamport 启发,并针对 Flink 执行模型量身定制。</p>    <h2>Barriers</h2>    <p>Flink 分布式快照的核心概念之一就是数据栅栏(barrier)。这些 barrier 被插入到数据流中,作为数据流的一部分和数据一起向下流动。Barrier 不会干扰正常数据,数据流严格有序。一个 barrier 把数据流分割成两部分:一部分进入到当前快照,另一部分进入下一个快照。每一个 barrier 都带有快照 ID,并且 barrier 之前的数据都进入了此快照。Barrier 不会干扰数据流处理,所以非常轻量。多个不同快照的多个 barrier 会在流中同时出现,即多个快照可能同时创建。</p>    <p style="text-align: center;"><img src="https://simg.open-open.com/show/324cba43fcbff34aa6c213c54f02e709.jpg"></p>    <p>Barrier 在数据源端插入,当 snapshot n 的 barrier 插入后,系统会记录当前 snapshot 位置值 n (用Sn表示)。例如,在 Apache Kafka 中,这个变量表示某个分区中最后一条数据的偏移量。这个位置值 Sn 会被发送到一个称为 checkpoint coordinator 的模块。(即 Flink 的 JobManager).</p>    <p>然后 barrier 继续往下流动,当一个 operator 从其输入流接收到所有标识 snapshot n 的 barrier 时,它会向其所有输出流插入一个标识 snapshot n 的 barrier。当 sink operator (DAG 流的终点)从其输入流接收到所有 barrier n 时,它向 the checkpoint coordinator 确认 snapshot n 已完成。当所有 sink 都确认了这个快照,快照就被标识为完成。</p>    <p style="text-align: center;"><img src="https://simg.open-open.com/show/e078cfc7848c6c08bef068268160321f.jpg"></p>    <p>接收超过一个输入流的 operator 需要基于 barrier 对齐(align)输入。参见上图:</p>    <ul>     <li> <p>operator 只要一接收到某个输入流的 barrier n,它就不能继续处理此数据流后续的数据,直到 operator 接收到其余流的 barrier n。否则会将属于 snapshot n 的数据和 snapshot n+1的搞混</p> </li>     <li> <p>barrier n 所属的数据流先不处理,从这些数据流中接收到的数据被放入接收缓存里(input buffer)</p> </li>     <li> <p>当从最后一个流中提取到 barrier n 时,operator 会发射出所有等待向后发送的数据,然后发射snapshot n 所属的 barrier</p> </li>     <li> <p>经过以上步骤,operator 恢复所有输入流数据的处理,优先处理输入缓存中的数据</p> </li>    </ul>    <h2>State</h2>    <p>operator 包含任何形式的状态,这些状态都必须包含在快照中。状态有很多种形式:</p>    <ul>     <li> <p>用户自定义状态:由 transformation 函数例如( map() 或者 filter())直接创建或者修改的状态。用户自定义状态可以是:转换函数中的 Java 对象的一个简单变量或者函数关联的 key/value 状态。</p> </li>     <li> <p>系统状态:这种状态是指作为 operator 计算中一部分缓存数据。典型例子就是: 窗口缓存(window buffers),系统收集窗口对应数据到其中,直到窗口计算和发射。</p> </li>    </ul>    <p>operator 在收到所有输入数据流中的 barrier 之后,在发射 barrier 到其输出流之前对其状态进行快照。此时,在 barrier 之前的数据对状态的更新已经完成,不会再依赖 barrier 之前数据。由于快照可能非常大,所以后端存储系统可配置。默认是存储到 JobManager 的内存中,但是对于生产系统,需要配置成一个可靠的分布式存储系统(例如 HDFS)。状态存储完成后,operator 会确认其 checkpoint 完成,发射出 barrier 到后续输出流。</p>    <p>快照现在包含了:</p>    <ul>     <li> <p>对于并行输入数据源:快照创建时数据流中的位置偏移</p> </li>     <li> <p>对于 operator:存储在快照中的状态指针</p> </li>    </ul>    <p style="text-align: center;"><img src="https://simg.open-open.com/show/b03dee2fe11b6197f3a5a577708c0f56.jpg"></p>    <h2>Exactly Once vs. At Least Once</h2>    <p>对齐操作可能会对流程序增加延迟。通常,这种额外的延迟在几毫秒的数量级,但是我们也遇到过延迟显著增加的异常情况。针对那些需要对所有输入都保持毫秒级的应用,Flink 提供了在 checkpoint 时关闭对齐的方法。当 operator 接收到一个 barrier 时,就会打一个快照,而不会等待其他 barrier。</p>    <p>跳过对齐操作使得即使在 barrier 到达时,Operator 依然继续处理输入。这就是说:operator 在 checkpoint n 创建之前,继续处理属于 checkpoint n+1 的数据。所以当异常恢复时,这部分数据就会重复,因为它们被包含在了 checkpoint n 中,同时也会在之后再次被处理。</p>    <p><em>注意:</em> 对齐操作只会发生在拥有多输入运算(join)或者多个输出的 operator(重分区、分流)的场景下。所以,对于自由 map(), flatmap(), fliter() 等的并行操作即使在至少一次的模式中仍然会保证严格一次。</p>    <h2>Asynchronous State Snapshots</h2>    <p>我们注意到上面描述的机制意味着当 operator 向后端存储快照时,会停止处理输入的数据。这种同步操作会在每次快照创建时引入延迟。</p>    <p>我们完全可以在存储快照时,让 operator 继续处理数据,让快照存储在后台异步运行。为了做到这一点,operator 必须能够生成一个后续修改不影响之前状态的状态对象。例如 RocksDB 中使用的写时复制( copy-on-write )类型的数据结构。</p>    <p>接收到输入的 barrier 时,operator 异步快照复制出的状态。然后立即发射 barrier 到输出流,继续正常的流处理。一旦后台异步快照完成,它就会向 checkpoint coordinator(JobManager)确认 checkpoint 完成。现在 checkpoint 完成的充分条件是:所有 sink 接收到了 barrier,所有有状态 operator 都确认完成了状态备份(可能会比 sink 接收到 barrier 晚)。</p>    <h2>Recovery</h2>    <p>在这种容错机制下的错误回复很明显:一旦遇到故障,Flink 选择最近一个完成的 checkpoint k。系统重新部署整个分布式数据流,重置所有 operator 的状态到 checkpoint k。数据源被置为从 Sk 位置读取。例如在 Apache Kafka 中,意味着让消费者从 Sk 处偏移开始读取。</p>    <p>如果是增量快照,operator 需要从最新的全量快照回复,然后对此状态进行一系列增量更新。</p>    <h2>Operator Snapshot Implementation</h2>    <p>当 operator 快照创建时有两部分操作:同步操作和异步操作。</p>    <p>operator 和后端存储将快照以 Java FutureTask 的方式提供。这个 task 包含了同步操作已经完成,异步操作还在等待的状态(state)。异步操作在后台线程中被执行。</p>    <p>完全同步的 operator 返回一个已经完成的 FutureTask 。如果异步操作需要执行,FutureTask 中的 run() 方法会被调用。</p>    <p>为了释放流和其他资源的消耗,可以取消这些 task。</p>    <p> </p>    <p>来自:https://segmentfault.com/a/1190000008129552</p>    <p> </p>    
 本文由用户 mthj8765 自行上传分享,仅供网友学习交流。所有权归原作者,若您的权利被侵害,请联系管理员。
 转载本站原创文章,请注明出处,并保留原始链接、图片水印。
 本站是一个以用户分享为主的开源技术平台,欢迎各类分享!
 本文地址:https://www.open-open.com/lib/view/open1484553716059.html
分布式系统 分布式/云计算/大数据 Apache Flink