Flink 经典课程四：Fault-tolerance_快照_状况

1.有状态的流打算

2.全局同等性快照

3.Flink的容错机制

4.Flink的状态管理

流打算

流打算是指有一个数据源可以持续不断地发送，同时有一个常驻程序运行代码，从数据源拿到一个后会进行处理，然后把结果输出到下贱。

分布式流打算

分布式流打算是指把输入流以某种办法进行一个划分，再利用多个分布式实例对流进行处理。

流打算中的状态

打算可以分成有状态和无状态两种，无状态的打算只须要处理单一事宜，有状态的打算须要记录并处理多个事宜。

举个大略的例子。
例如一个事宜由事宜ID和事宜值两部分组成，如果处理逻辑是每拿到一个事宜，都解析并输出它的事宜值，那么这便是一个无状态的打算；相反，如果每拿到一个状态，解析它的值出来后，须要和前一个事宜值进行比较，比前一个事宜值大的时候才把它进行输出，这便是一个有状态的打算。

流打算中的状态有很多种。
比如在去重的场景下，会记录所有的主键；又或者在窗口打算里，已经进入窗口还没触发的数据，这也是流打算的状态；在机器学习/深度学习场景里，演习的模型及参数数据都是流打算的状态。

全局同等性快照是可以用来给分布式系统做备份和故障规复的机制。

全局快照

什么是全局快照

全局快照首先是一个分布式运用，它有多个进程分布在多个做事器上；其次，它在运用内部有自己的处理逻辑和状态；第三，运用间是可以相互通信的；第四，在这种分布式的运用，有内部状态，硬件可以通信的情形下，某一时候的全局状态，就叫做全局的快照。

为什么须要全局快照

第一，用它来做检讨点，可以定期对全局状态做备份，当运用程序故障时，就可以拿来规复；第二，做去世锁检测，进行快照后当前的程序连续运行，然后可以对快照进行剖析，看运用程序是不是存在去世锁状态，如果是就可以进行相应的处理。

全局快照举例

下图为分布式系统中全局快照的示例。

P1和P2是两个进程，它们之间有发送的管道，分别是C12和C21。
对付 P1进程来说， C12是它发送的管道，称作output channel； C21是它吸收的管道，称作 input channel。

除了管道，每个进程都有一个本地的状态。
比如说P1和P2每个进程的内存里都有XYZ三个变量和相应的值。
那么 P1和P2进程确当地状态和它们之间发送的管道状态，便是一个初始的全局状态，也可称为全局快照。

假设P1给P2发了一条，让P2把x的状态值从4改为7，但是这个在管道中，还没到达P2。
这个状态也是一个全局快照。

再接下来，P2收到了P1的，但是还没有处理，这个状态也是一个全局快照。

末了接到的P2把本地的X的值从4改为7，这也是一个全局快照。

以是当有事宜发生的时候，全局的状态就会发生改变。
事宜包括进程发送、进程吸收和进程修正自己的状态。

2.全局同等性快照

如果说有两个事宜，a和b，在绝对韶光下，如果a发生在b之前，且b被包含在快照当中，那么则a也被包含在快照当中。
知足这个条件的全局快照，就称为全局同等性快照。

2.1 全局同等性快照的实现方法

时钟同步并不能实现全局同等性快照；全局同步虽然可以实现，但是它的缺陷也非常明显，它会让所有运用程序都停下来，会影响全局的性能。

3.异步全局同等性快照算法 – Chandy-Lamport

异步全局同等性快照算法Chandy-Lamport可以在不影响运用程序运行的条件下，实现全局同等性快照。

Chandy-Lamport的系统哀求有以下几点：

同时，Chandy-Lamport算法可以实行还有一个条件条件：有序且不重复，并且可靠性可保障。

3.1 Chandy-Lamport算法流程

Chandy-Lamport的算法流程紧张分为三个部分：发起快照、分布式的实行快照和终止快照。

发起快照

任意进程都可以发起快照。
如下图所示，当由P1发起快照的时候，第一步须要记录本地的状态，也便是对本地进行快照，然后急速向它所有 output channel发送一个marker，这中间是没有韶光间隙的。
marker是一个分外的，它不同于运用之间通报的。

发出Marker后，P1就会开始记录所有input channel的，也便是图示C21管道的。

分布式的实行快照

如下图，先假定当 Pi吸收到来自Cki的marker，也便是Pk发给Pi的marker。
可以分两种情形来看：

第一种情形：这个是Pi收到的第一个来自其它管道的marker，它会先记录一下本地的状态，再把 C12管道记为空，也便是说后续再从 P1发，就不包含在这次快照里了，与此同时急速向它所有output channel发送marker。
末了开始记录来自除Cki之外的所有input channel的。

上面提到Cki不包含在实时快照里，但是实时还是会发生，以是第二种情形是，如果此前Pi已经吸收过marker，它会停滞记录 Cki，同时会将此前记录的所有Cki作为Cki在本次快照中的终极状态来保存。

终止快照

终止快照的条件有两个：

当快照终止，快照网络器 (Central Server) 就开始网络每一个部分的快照去形玉成局同等性快照了。

示例展示

不才图的例子里，一些状态是在内部发生的，比如A，它跟其它进程没有交互。
内部状态便是 P1发给自己，可以将A认为是C11=[A->]。

Chandy-Lamport全局同等性快照的算法是怎么实行的呢？

假设从p1来发起快照，它发起快照时，首先对本地的状态进行快照，称之为S1，然后急速向它所有的output channel，即P2和P3，分别发marker，然后再去记录它所有input channel的，即来自P2和P3及自身的。

图例所示，纵轴是绝对韶光，按照绝对韶光来看，为什么P3和P2收到marker会有韶光差呢？由于如果这是一个真实的物理环境里的分布式进程，不同节点之间的网络状况是不一样的，这种情形会导致投递韶光存在差异。

P3先收到marker，且是它吸收到的第一个marker。
吸收到后，它首先会对本地状态进行快照，然后把 C13管道的标记成 close，与此同时开始向它所有的output channel发送 marker，末了它会把来自除了C13之外的所有input channel的开始进行记录。

吸收到P3发出的marker信息的是P1，但这不是它吸收的第一个marker，它会把来自C31 channel的管道急速关闭，并且把当前的记录做这个channel的快照，后续再吸收到来自P3的，就不会更新在这次的快照状态里了。

接下来P2吸收到来自P3的，这是它接到的第一个marker。
吸收到后，它首先对本地状态进行快照，然后把 C32管道的标记成 close，与此同时开始向它所有的output channel发送 marker，末了它会把来自除了C32之外的所有input channel的开始进行记录。

再来看P2吸收到来自P1的，这不是P2吸收到的第一个marker，以是它会把所有的 input channel全部关闭，并且记录channel的状态。

接下来看P1吸收到来自P2的，这也不是它吸收的第一个。
那么它就会把所有的input channel关闭，并把记录的作为状态。
那么这里面有两个状态，一个是C11，即自己发给自己的；一个是C21，是P2里H发给P1D的。

末了一个韶光点，P3吸收到来自P2的，这也不是它收到的第一个，操作跟上面先容的一样。
在这期间P3本地有一个事宜J，它也会把J作为它的状态。

当所有进程都记录了本地状态，而且每一个进程的所有输入管道都已经关闭了，那么全局同等性快照就结束了，也便是对过去韶光点的全局性的状态记录完成了。

3.3 Chandy-Lamport与 Flink之间的关系

Flink 是分布式系统，以是 Flink 会采取全局同等性快照的办法形成检讨点，来支持故障规复。
Flink的异步全局同等性快照算法跟Chandy-Lamport算法的差异紧张有以下几点：

容错，便是规复到出错前的状态。
流打算容错同等性担保有三种，分别是：Exactly once，At least once，At most once。

端到真个Exactly once

Exactly once的意思是，作业结果总是精确的，但是很可能产出多次；以是它的哀求是须要有可重放的source。
端到真个Exactly once，是指作业结果精确且只会被产出一次，它的哀求除了有可重放的source外，还哀求有事务型的sink和可以吸收幂等的产出结果。

Flink的状态容错

很多场景都会哀求在Exactly once的语义，即处理且仅处理一次。
如何确保语义呢？

大略场景的 Exactly Once 容错方法

大略场景的做法如下图，方法便是，记录本地状态并且把 source的offset，即 Event log的位置记录下来就好了。

分布式场景的状态容错

如果是分布式场景，我们须要在不中断运算的条件下对多个拥有本地状态的算子产生全局同等性快照。
Flink 分布式场景的作业拓扑比较分外，它是有向无环并且是弱联通图，可以采取裁剪的Chandy-Lamport，也便是只记录所有输入的offset和各个算子状态，并依赖rewindable source（可回溯的source，即可以通过offset读取比较早一点韶光点），从而不须要存储channel的状态，这在存在聚合 (aggregation）逻辑的情形下可以节省大量的存储空间。

末了做规复，规复便是把数据源的位置重新设定，然后每一个算子都从检讨点规复状态。

3.Flink 的分布式快照方法

首先在源数据流里插入Checkpoint barrier，也便是上文提到的Chandy-Lamport算法里的marker message，不同的Checkpoint barrier会把流自然地切分多个段，每个段都包含了Checkpoint的数据；

Flink 里有一个全局的Coordinator，它不像Chandy-Lamport对任意一个进程都可以发起快照，这个集中式的 Coordinator会把Checkpoint barrier注入到每个source里，然后启动快照。
当每个节点收到barrier后，由于 Flink 里面它不存储 Channel state，以是它只需存储本地的状态就好。

在做完了Checkpoint 后，每个算子的每个并发都会向Coordinator发送一个确认，当所有任务的确认都被Checkpoint Coordinator吸收，快照就结束了。

4.流程演示

见下图示，假设Checkpoint N 被注入到 source里，这时source会先把它正在处理分区的offset记录下来。

随着韶光的流逝，它会把Checkpoint barrier发送到两个并发的下贱，当barrier分别到达两个并发，这两个并发会分别把它们本地的状态都记录在Checkpoint 的里：

末了barrier到达终极的subtask，快照就完成了。

这是比较大略的场景演示，每个算子只有单流的输入，再来看下图比较繁芜的场景，算子有多流输入的情形。

当算子有多个输入，须要把Barrier 对齐。
怎么把Barrier对齐呢？如下图所示，在左侧原来的状态下，当个中一条barrier到达，另一条barrier命令上有的barrier还在管道中没有到达，这时会在担保Exactly once的情形下，把先到达的流直接壅塞掉，然后等待另一条流的数据处理。
等到其余一条流也到达了，会把之前的流unblock，同时把barrier发送到算子。

在这个过程中，壅塞掉个中一条流的浸染是，会让它产生反压。
Barrier 对齐会导致反压和停息operator的数据处理。

如果不在对齐过程中壅塞已收到barrier的数据管道，数据持续不断流进来，那么属于下个Checkpoint的数据被包含在当前的Checkpoint里，如果一旦发生故障规复后，由于source会被rewind，部分数据会有重复处理，这便是at-least-once。
如果能吸收at-least-once，那么可以选择其他可以避免barrier对齐带来的副浸染。
其余也可以通过异步快照来只管即便减少任务停顿并支持多个Checkpoint同时进行。

5.快照触发

本地快照同步上传到系统须要state Copy-on-write的机制。

如果对元数据信息做了快照之后数据处理规复了，在上传数据的过程中如何担保规复的运用程序逻辑不会修正正在上传的数据呢？实际上不同状态存储后真个处理是不一样的，Heap backend会触发数据的copy-on-write，而对付RocksDB backend来说LSM的特性可以担保已经快照的数据不会被修正。

四、Flink 的状态管理

1.Flink 状态管理

首先须要去定义一个状态，不才图的例子里，先定义一个Value state。

在定义的状态的时候，须要给出以下的几个信息：

2.Flink 状态后端

又叫state backend，Flink状态后端有两种；

第一种，JVM Heap，它里面的数据因此Java工具形式存在的，读写也因此工具形式去完成的，以是速率很快。
但是也存在两个弊端：第一个弊端，以工具办法存储所需的空间是磁盘上序列化压缩后的数据大小的很多倍，以是占用的内存空间很大；第二个弊端，虽然读写不用做序列化，但是在形成snapshot时须要做序列化，以是它的异步snapshot过程会比较慢。

本文为阿里云原创内容，未经许可不得转载。