分布式数据流的轻量级异步快照

本文章学习：

Lightweight Asynchronous Snapshots for Distributed Dataﬂows

分布式状态流处理其被用于云部署并执行大规模连续计算，致力于高吞吐和低延迟。这依靠于周期性全局状态快照，他有两个缺点：

由于影响了数据的摄入从而拖延了整体计算。 2.他们坚持了存储所有传输和算子的状态，从而导致更大的快照。下面我们介绍Asynchronous Barrier Snapshotting (ABS，异步屏障快照)。

简介

现有的所有分布式快照算法都把在通道中传输的记录或整个执行图中未处理的消息作为快照状态的一部分，这通常都不是必须的。而对于ABS来说，并不会停止流的操作，仅仅有一点小的开销。全局快照从理论上反映了执行图的整体状态或者其操作的一个特定实例的一个可能状态。

Apache Flink

当一个用户需要执行一个应用程序时，所有的流数据运算符都被编译为一个执行图，原则上是一个有向图G = (T,E) ，其中顶点T表示任务，边E表示数据通道。运算符的每个实例都封装在各自的任务(subtask)上。M表示在并行执行期间由任务传输的所有记录的集合。每一个t∈T都是封装运算符实例的独立执行并且由以下内容组成：

一组输入和输出管道 It,Ot ⊆ E;
一个运算符状态st ;
一个用户自定义函数(UDF)ft; 对于执行过程来说，每一个task都需要消耗输入记录，更新自己的状态，并根据UDF生成新的记录。换种说法是，对于每条记录 r∈ M被任务t∈T接收，并生成一个新的状态和一组根据UDF ft: 产生的新的输出D⊆M。

ABS

我们定义了执行图G = (T,E)的一个全局快照G*=(T*, E*)作为所有任务和边的状态的集合，T由运算符状态st∈T组成，其中t∈T，E由通道状态e∈E组成，其中e∈E。

非环状执行图的异步屏障快照

1: upon event <Init | input channels, output channels, fun, init state> do // 事件发生<触发操作 | 对象>执行
2:    blocked inputs:= 0; //阻塞管道数初始化0
3:    inputs:= input channels; //初始化input为所有input管道数
4:    outputs:= output channels; UDF := fun; //初始化output为所有输出管道数，初始化UDF
5: 
6: upon event <receive | input,<barrier>> do //从某个input接收到barrier
7:     if input != Nil then //当该input不是source时
8:         blocked inputs := blocked inputs ∪ {input}; //将已经收到barrier的管道并入 blocked inputs并阻塞，其他管道继续接收。
9:         trigger <block | input>; //触发阻塞barrier已经到达的管道
10:   if blocked_inputs = inputs then //已收到barrier的管道数等于所有input端的管道数，即所有barrier已经到达
11:       blocked_inputs:= 0; //置0
12:       broadcast <send | outputs,<barrier>>; //向下游广播barrier
13:       trigger<snapshot | state>; //触发快照，保存算子state
14:       for each inputs as input 
15:           trigger <unblock | input>; //解锁所有input管道
16: 
17: 
18: upon event <receive | input, msg> do//接受到正常消息时
19:     {state', out_records} := UDF(msg, state); //计算新的状态值和输出记录
20:     state := state' ; //更新状态值
21:     for each out_records as {output,out_record} 
22:         trigger <send | output, out_record>; //向下游发送经算子计算后的输出记录。

当通道被阻塞时，所有消息被缓冲，不会被传递，直到取消阻塞。
task可以在诸如block, unblock and send messages等通道组件上触发操作，所有输出通道都支持broadcasting messages。
在source被注入的消息被解析为一个"Nil"输入通道。

中央协调员(JobManager中的一个进程)周期性的向所有source注入barrier。当一个source接受到barrier时，它会建立当前状态的快照并向下游广播。当非源算子接受到barrier时，它会阻塞该input channel，直到所有input都收到barrier后，该算子会保存当前状态的快照并向下游广播barrier。然后所有被block的input通道解除阻塞。完整的全局快照 G=(T, E) 将独自由所有运算符的状态集T组成，其中E* = 0。

环状数据流的ABS

在有向循环图中使用上面的算法时，因为算子会无期限的等待接受所有输入管道的barrier，将不会停止，会导致死锁。下面是适用于环状拓扑的ABS：

1: upon event <Init | input_channels, backedge_channels, output_channels, fun, init_state> do 
2:     state:= init_state; marked := 0; 
3:     inputs:= input_channels; logging:= False; 
4:     outputs:= output_channels; UDF := fun; 
5:     loop_inputs:= backedge_channels; 
6:     state_copy:= Nil; backup_log:= []; 
7: 
8: upon event <receive | input, <barrier>> do 
9:     marked := marked∪{input}; 
10:     regular := inputs\loop_inputs; 
11:     if input != Nil AND input / ∈loop inputs then 
12:         trigger <block | input>; 
13:     if ¬logging AND marked = regular then 
14:         state_copy:= state;logging:= True; 
15:         broadcast <send | outputs, <barrier>>; 
16:         for each inputs as input 
17:             trigger <unblock | input>; 
18: 
19:     if marked = input_channels then 
20:         trigger <snapshot | {state_copy, backup_log}>; 
21:         marked := 0;logging:= False; 
22:         state_copy:= Nil;backup_log:= []; 
23: 
24: upon event <receive|input, msg> do 
25:     if logging AND node∈loop_inputs then 
26:         backup_log:= backup_log:: [input]; 
27:     {state',out_records}:= UDF(msg,state); 
28:     state:= state'; 
29:     for each out_records as{output,out_record} 
30:         trigger <send | output, out_record>;

实现

在当前的实现中，blocked channels的所有输入数据被存储在磁盘上，而不是保存在内存中，以提高可伸缩性。
为了区分运算符状态和数据，我们引入了一个显式的OperatorState接口用于查询和更新状态。并且为Apache Flink的一些有状态的操作符(例如基于源的偏移量和聚合)提供了OperatorState实现。

结果

ABS与全局同步快照算法比较

上述用于评估的执行拓扑图由六个不同运算符组成，并行度等于集群节点数目。source生成总计10亿条记录，在source实例间均匀分布。拓扑图中运算符的状态为每键(per-key)聚合和源偏移量。

在图2中, 我们描述了两个算法对基线 (Baseline) 的运行时影响。当快照的间隔较小时, 同步快照的性能影响尤为明显。这是由于系统花费更多的时间不处理任何数据, 以获取全局快照。ABS 对运行时的影响要低得多, 因为它连续运行而不阻塞总体执行, 同时保持相当稳定的吞吐量。对于更大的快照间隔, 同步算法的影响不那么显著, 因为它在我们的实验中以1-2 秒的时间运行, 同时让系统在其其余执行过程中以正常的吞吐量运转。

在图3中, 我们比较了运行 ABS 的拓扑的可伸缩性, 与基线的3秒快照间隔(无容错)相比。很明显, 基线作业和 ABS 都实现了线性可伸缩性。

GQBBBB / GQBBBB.github.io