若是让你去数羊 -- 谈分布式异步快照原理

当你第一次看断背山的时候,印象最深入的是什么? 有没有多是他们数羊的片断?? 哈哈,虽然确定不是,可是今天的问题和数羊有关系. 若是一群羊都在跑, 你怎么数的清? (前提是没有标记)

我先表扬本身一下, 由于在我看来,数羊是对于下面这个问题的一个绝妙的比喻.
可能你已经想到一个方法: 那就是给整个羊群照一张照片,而后本身拿着照片回家慢慢数. 这个方法用术语来表示就是你获得了一个全局快照(global snapshot).经过这个快照,你保留了一个能够恢复能够处理的稳定状态.
然而问题又来了,羊太多,一张照片照不全该怎么办?你怎么保证同时照几张照片呢?

若是你说把羊所有打晕再数, 那我如今就打晕你.

前言

好久不更新了,最近在公司过了一段读论文的日子,为了下一version项目的实时计算部分阅读了不少流计算资料. 包括Apache著名的四大流计算框架. Apache Storm, Spark streaming,Apache Flink,Samza. 以及google刚刚联合apache基金会成功孵化出的大数据处理统一编程模型 Apache Beam( 原型为Google的Dataflow.)

如下挑出一个有意思的点来分享. 这个是Apache Flink 消息传递机制保证和容错机制中确保每一个消息被处理且仅被处理一次(因此不会出现消息丢失,同事又不会由于容错中消息重发而致使消息重复)的依赖原理.

背景

无统一时钟

在计算机世界中,各个机器很难有一个统一的时钟来有统一的时间表示. 可能你会想咱们人类世界中就能够保证你们对时间的认识一致.好比你和小明约定”咱们明早8点的时候同时开始看某个电影”,你说的8点和小明说的8点咱们都认为是一个时刻,即大家是能够实现”同时”的.

这个实际上是由于咱们统一按照格林尼治的时间来作了一个统一. 可是让分布式系统中的每一个机器都作到这样是很难的,或者说有一个总的master强制统一这个”同时”的开销是很大的.

若是没有这个”同时”,咱们就没法得到一个全局的状态(global state).

全局状态 & 稳定特征感应

而分布式系统中的不少问题均可以化为想要获得全局状态的问题.
好比,你想知道系统是否达到了稳定特征(stable property),譬如死锁问题,你须要这种稳定的状态来探测到系统已经发生死锁.

稳定性的定义是这种状态会永久的持续下去,你能够想象在无阻力环境中的匀速运动或者静止状态.

稳定性感应能够用来作全局快照. 当稳定以后,好比羊群都中止移动或者同速同向平移,你能够在不一样时刻照不少照片来合成一张”同时”的全局大照片.

同步快照 & 分布式异步快照

你想要在同一时刻保存系统状态,这个便是同步快照.

分布式异步快照的主要思想是你经过协调,使得他们并非在同一时刻保存快照,可是快照可以反应出系统数据的处理状态. 即一种”逻辑同步”, 这个是基于Chandy-Lamport算法,严谨地被证实在论文中.(reference中可查)

流式计算中的exactly-once 语意容错机制

消息传输保障机制分为三种:

1.at most once

一条消息通过系统，由这条消息产生的后续tuple在各个处理节点最多会被处理一次，含义就是，出现故障时，不保证这条消息本来应该涉及的全部处理节点计算都顺利完成。

2.at least once

一条消息通过系统，由这条消息产生的后续tuple在各个处理节点至少会被处理一次，含义就是，出现故障时，系统可以识别并进行tuple重发，可是没办法判断是否以前该元组被成功处理完成了，所以可能会有重复处理的状况，对于某些改变外部状态的场景，会形成脏数据。

3.exactly once

一条消息通过系统，不论是否发生故障，由其产生的后续tuple，在全部处理节点上有且仅会被处理一次，这是最理想的状况，即便出现故障，也能符合正确的业务预期，但通常会带来比较大的性能开销。

由于作到exactly-once 有相对较大的性能开销,而且不是幂等的计算所必须,因此并不是全部的流计算框架作到了这一点.

相对于micro-batch底层实现的spark streaming,Apache Flink 便使用了分布式快照和检查点(checkpointing)机制来实现了exactly-once 的容错级别.

Flink 进行周期性的全局快照(periodic global state)保存,从而在出现系统failure的时候,只要从上一次保存成功的全局快照中恢复每一个节点的恢复状态,而后再使源数据节点从相应快照标记源数据节点从新开始处理便可恢复无措运行状态(Kafka能够作到这一点).

Asynchronous Barrier Snapshot(ABS)

同步快照有如下两种潜在的问题:

1. 须要全部节点中止工做,即暂停了整个计算,这个必然影响数据处理效率和时效性.
2. 须要保存全部节点/操做中的状态以及全部在传输中的数据(好比storm中的tuples),这个会消费大量的存储空间.

轻量级的异步栅栏快照能够用来为数据流引擎提供容错机制,同时减少的存储空间需求.
由于ABS只须要保存一个无环拓扑中每一个操做节点的处理状态(operator states).
Apache Flink 就使用了ABS的机制.

当运行能够分为几个阶段的时候,快照是能够不包含每一个task节点间的通道状态的.

Stages divide the injected data streams and all associated into a series of possible executions where all prior inputs and generated outputs have been fully processed.
The set of operator states at the end of a stage reflects the whole execution history, therefore, it can be solely used for a snapshot. The core idea behind our
algorithm is to create identical snapshots with staged
snapshotting while keeping a continuous data ingestion.

这段话太经典.

能够经过切分源数据来划分阶段,每一个集合的源数据也表明了其所须要的计算. 当上一个集合的输入数据以及输出都被彻底处理,就表明上一个阶段结束了.

因此当一个阶段结束时,操做节点的状态能够表明这整个个运行历史,从而快照能够仅仅依赖于operator states.

这些阶段能够经过周期性的在源头出插入一些栅栏(barrier)来划分. 这些栅栏起到了阶段的标记做用,而后跟着数据流经过整个数据处理pipeline,知道系统的sinks.

全局状态在这个过程当中,被增量地构建, 即当每一个处理tast接收到对应id的栅栏的时候对本身的状态进行快照,而每一个节点异步的快照共同组成了一个阶段(stage)

详细的过程能够见下图, Source1,2,3,4 在接到Master的checkpointing消息时,

保存本身的消息消费状态,而后释放一个barrier(包含一个id标记).
当以后的task节点在接收到barrier时,中止处理下一条数据,立刻对本身的状态进行快照而且持久化存储, 而且记录此次状态的id, 快照保存以后继续输出barrier并恢复数据梳理流程.

当数据sinks收到全部barriers而且进行自身状态保存以后,也进行ack的checkpointing.

论文算法以下:

看到这里,你会数羊了吗?