059.尚硅谷_Flink-容错机制_检查点算法

然后接下来我们来具体的分析一下，在flink里边它的拆矿的这个检查点具体实现的算法到底是什么样的啊，有同学可能会觉得比较奇怪，这里边这个检查点还有什么这个算法吗？这不就是所有的数据来了之后存一份不就完了吗？而且之前我们也已经说了，呃，就是你你现在的要求是每一个数据处理完，就是要求同一个数据对吧？输入数据五处理完成之后，所有的任务都处理完成之后啊，去做一个这个存盘，这就可以了嘛，诶这里边的问题就来了。问题就在于，那就是你这里边要求，如果说这个五全部的任务都是五处理完之后去做一个存盘的话，这个任务他怎么知道是五处理完成之后要做一个存盘呢？难道他要检测一下当前的这个数据等于五吗？那万一这个数据还有重复的数据呢？这个数据上有有标签吗？没有标签啊，你怎么样去判断当前就真的是要做一个存盘了呢？

啊，那有同学可能说，哎，那就你你还是不要这么麻烦吧，我们直接简单实现，还是所有的这个都直接暂停下来，然后直接保存就完事了嘛，啊这种暂停的这种方式啊，就是直接停止一切的这种方式啊，这是不推荐的，因为我们自然能想到你这个，呃，首先就是我们前面说的啊，你要保存的时候，那是不是相当于还得有上下文，它的一些就是当前处理到哪里了，对吧，你当前的这个上下文还得做保存，这个其实要保存的东西还是比较多的，另外还有一个问题就是说大家想想，你如果要是就是保存一个检查点的时候，我就要求当前所有任务都全部都停下，因为大家想你如果现在就是至少是一部分任务要停下来等别的，如果要有有有任务没有，呃，处理完的话，你要等他对不对啊，你你要等到所有的这个任务都对齐的时候，你才能一下子喊茄子嘛，就像我们这个拍合照的时候，有些同学。

可能就会觉得这个不喜欢拍合照对吧？呃，就因为就是拍合照的时候总要等人，你总得等到大家手头事情全做完，然后站到一起，然后才能123茄子拍一张合照，这种方式其实是，呃，这效率上是很低的，对吧？就且且不说你要保存你当前的这个工作进度啊，先做一个这个上下文的保存，那另外就是说从效率上来讲，我们大家都在等，就很多人下去之后是在漫长的等待过程当中，那对于这个大数据处理应用而言，我们当前如果说这个延迟是因为要保证结果的正确性，就像前面我们讲的这个watermark设置一个延迟，对吧，或者说这个窗口多等一会儿，等待迟到数据，这个我还是可以接受的，因为我要达到一定的效果嘛，你现在就是为了做一个快照保存，然后你就要所有的人都暂停，现在的正事不能干，对吧，然后就要拍张照片，这显然是不可接受的。

那所以说在这个flink的实现里边，我们实现的呢，还是不要暂停整个任务，而是怎么样？呃，就是把每一个，就是我要判断当前是哪一个数据处理完了之后，我要做一个快照的保存。啊，那有同学可能就想了，那这个，呃，这个数据就不好去定义啊，你怎么样去表示这个当前数据已经处理完了呢？当前数据确实是没有办法给它加一个限制，说诶就是这个数据对吧，你就找这个数据，收到这个数据处理完了之后你就保存，但是我们可以怎么样呢？诶之前大家就就想到了，我们讲那个water mark的时候，不是说可以在数据后边诶插入周期性的去插入一个water mark，表示我们当前的时间进展吗？现在也是有类似的思路，我可以怎么样呢？我插入一个标记表示什么呢？哎，就是前面的数据都处理完了，现在我该做一次拆框保存了啊，所以说就相当于可以引入这样的一个标记机制，对吧？啊，那具体来讲的话，底层弗林克里边的实现，它是一个所谓的基于昌迪兰波特算法的一个分布式快照这样的一个机制，那呃，它的整体的思路是什么呢？就是把检查点的保存和这个数据的处理过程啊直接分离开，就是我在保存的时候不需要暂停整个应用，谁该保存的时候你去做一个保存就可以了，那那整体来讲，这个如果对比大家拍这个照片，那就相当于是什么呢？就相当于我设置这样的一个标志，对吧？啊，就是大家每个人还是你该做手头的事儿，直接去做就可以了，然后我设置一个节点，比方说link学习完的时。

之后大家，哎，这个到楼下去拍一张照片，那如果要是这样的话，所有人都可以知道，我现在就就学嘛，进度可能不一样，有些同学学完了，学完之后我就这个下去之后自己单独拍一张照片就完事了，那其他同学也是做完这个学，呃，弗林格的学习之后，到下面自己单独拍一张照片啊，那最后我们怎么样得到一张合照呢？哎，最后把每个人的拍的那张照片再P起来，拼到一张合照里边不就完成了吗？

哎，所以说呃，这个我们在现实当中一般不会这么去干，那是因为P图拼，拼接合照这个工作量有点大对吧？但是你想对于这个计算机系统而言，拼接快照，拼接我们的这个每个任务，它的那个保存的那个数据，柴的一部分的那个数据啊呃，这个工作量很大吗？一点都不大对吧，直接把它拼起来就完事了嘛，所以用这种方式就可以极大的提高系统的性能，就不会让当前的checkpoint的操作对系统性性能造成比较大的影响。那接下来我们给大家说一说，就是具体的一个实现，前面已经说过我们的一个思路其实是什么呢？呃，别别看这个名头有点有点夸张啊，昌吉兰伯特这个分布式快照算法啊，那其实它的实现机制就是刚才我们说的，要在某个数据后边插入一个标记位，对吧？来告诉当前的任务，说你接收到前面那个数据，处理完了之后，假如后边你紧接着接收到这个标记了，你就做一个快照保存，做一个存盘，那所以这一个标记位叫做什么呢？叫做检查点分界线啊，英文名字叫做checkpoint barrier啊，有些时候翻译成分界线，有些时候可能翻译成这个，大家知道barrier有有篱笆屏障的那个意思，对吧？有时候翻译成这个检查点屏障啊，检查点那个分割符啊，这个怎么翻译都可以，它的主要的目的是什么呢？就是把我们数据相当于前后分开，对吧，告诉任。

说好之前的数据都处理完了，现在保存一份状态之后的数据呢？诶，那当前这个状态里面就不包括对吧，之后的数据，假如说呃，这个发挂了的话，之后的数据我恢恢复到这个状态之后的数据就全部都要重新处理一遍，之前的数据就不用重新处理了，这就是所谓这个分界线barrier的概念啊，它其实就跟这个watermark有点类似啊，就相当于就是说在这个数据流里边再插入一个特殊的数据结构。之前的呢，之前的数据它就会包在它的那个状态的更改啊，就都会包含在我们当前这个分界线表示的那个保存的状态里，对吧，那它之后的那些数据呢？啊，那就就只能包含在之后的那个检查点里了，当前这个状态就有。啊，这就是所谓的barrier的一个引入啊，那如果这么说的话，可能我们发现这也简单呀，对吧，这就相当于在这个处理的过程当中啊。

处理的这个流失计算的过程当中，再插一个这个分界线嘛，呃，你就相当于来了数据之后啊，那那这个分界线怎么样去去插进去呢？啊，那这就是我们前面说的是drop manager job manager有一个功能，它就是来控制这个检查点啊，进行保存的这个调度计算对吧？呃，调度处理所以说他那边会发起一个指令说诶那个所有的task manager注意了，现在我要去做一个检查点啊，那这个时候他的这个指令是发给谁呢？注意他不是发给所有的任务。他只发给，因为大家想是不是这里边我们的数据都是从往下一点一点流动的呀，因为我们说这个数据流嘛，对吧，所以这里边这个manager很聪明，他控制这个分界线的时候，他从源头这边直接通知给这个任务，然后让务把这个bar到自己当前处理完的这个数据之后。

然后接下来它往下游传递数据，是不是大家看这个就按照顺序，就保持这个顺序，直接顺这个往下一个一个传递了啊，就像是我们数据处理完了之后，接收到它的时候就去做保存一样，那后续任务是不是就相当于如果前面我们这个数据是五的话，后面的任务接收到的顺序也是这样对吧？前面的四和五处理完成之后收到了Barry，然后我就做一个存盘保存，然后后面的六和七就没有存到当前的这个状态里。如果后边这里挂了的话，我从这个状态恢复出来，那就是五处理完之后的状态，对吧，六和七要做一个重放。啊啊，那那大家会想到这个看起来就没什么算法可言啊，这个非常简单啊，不就这么做吗？那你就按照顺序一个一个处理就完了吗？哎，问题就又来了，我们这里给大家说的是就只有一个并行任务往下传递的过程，对吧？就像之前我们给大家讲watermark传递规则一样。

戴瑞是不是也应该得有一个传递规则呀？哎，这里边就真正涉及到了我们所谓的算法了，因为你看现在我们要如果有多个病行子任务的话，就会涉及到一个问题，我这里边的这个barriry，首先Barry怎么样朝下游传递啊，那有同学可能想到了广播嘛，对吧，就像我们那watermark一样，你既然表示之前的那个数据都处理完了，现在该做存盘了，那当然是通知下游到这儿的位置都都应该存盘了，对吧？诶，但是下游任务大家注意啊，是不是还会接收到不同的上游任务发来的bar瑞啊，那就有一个问题了，我下游任务到底是按照接收到谁的bar瑞作为标准来做这个存盘呢？啊，所以这里边就有一个有一个所谓的算法调整的一个策略了，啊，其实我们一想也就能想到，像之前我们做watermark处理的时候，它是要等这个所有的分区的那个water都涨到某一个值的时候，也就是找那个最小值对吧，都已经达到这个时间的时候，我的时间才推进到当前的这个状态，那同样现在这个bar瑞尔是不是相当于也应该是我要等上游所有任务的那个barrier都到齐的时候。

这才表示上游的所有数据都已经处理完了，对吧？哎，那现在是不是我才可以去做保存了呀啊，所以相当于我这里边就要收集上游所有的barrier，那这样的一个检查点算法啊，具体的这个检查点算法其实就叫做所谓的barrier对齐啊，接下来给大家再这个具体的流程当中，把这个再顺一遍，大家看一看，从图上看可能会看的更加清晰一点啊，那首先还是我们这个例子啊，就是奇偶数求和的这个例子，还是自然数流，1234567啊，一个一个按照顺序去输入，现在就复杂一点了，之前我们是一条流，现在我们有两条流了，这两条流呢？啊，数据都一样啊，都是1234567，然后我们就简单来分区，分的话就是上面这个叫蓝流，下面这个叫黄流吧，呃，那蓝流这里边大家看一下现在这个状态是什么样啊状态诶这里边有一个SS有两个S任务对吧，S1这里边。

读的是蓝流的这个数据，SS2这里呢，读的是这个，呃，黄流的数据，或者说大家可以认为这是两条流，也可以认为是一条流并行的这个SS任务对吧？哎，我同时在读数据而已，这个都是可以的啊，效果是一样的，然后我们看一下现在的状态，现在的状态是，呃，这个蓝流这里边读到了三，对吧，大家看这这里边状态偏移量是三嘛，S任务偏移量是三啊，黄流这里呢，读到了四，偏量是四，所以就是SS这边该读的都已经读完了，这蓝三皇四都经读完了，然后接下来后边的这个处理效果呢？哎，是这里边这个偶数求和，它的状态是二，为什么是二呢？呃，它就是246各种叠加吗？所以这两个流里边不都是有有这个1234567吗？所以这里边的二就应该只有一个二了，它是。

黄牛里边的二对吧，因为大家看这里面它输出了一个黄二吧，就是它输出的就是当前的那个萨姆结果嘛，直接求和的那个结果，但是这个这个二呢，还没有输出到think，没有写到外部系统对吧？这个还在路上，当前的这个状态是这个黄二引起的，然后蓝二呢还在路上的，然后面的这个黄四呢，也还在路上对吧？呃，这个还都没处理完呢啊，所以它的状态就只有一个二的叠加，那下面这个奇数求和呢，奇数求和是五五的，这个特点是什么呢？诶我们看这里它输出的这个效果啊，输出的时候大家知道这个奇数求和是135去叠加嘛，135去叠加，它其实一开始应该收到的只有一个一，对不对，一开始应该是只有一个一，然后后边呢，诶，可能会又来一个一，是不是加起来就变成二了啊，所以大家看这里边它输出了一个蓝二，蓝二表示什么呢？是后边的这个蓝一来了之后。

跟之前的状态叠加得到的这个结果引发的一个输出，哎，那所以前面理论上应该还有一个黄，黄一对吧？呃，那这是蓝一来的时候输出了这个二嘛，那黄一来的时候应该输出一个一啊，那这个一去哪儿了呢？一这里既然没有，那当然是已经写入到外部系统了，Think系统里面去了，对吧？啊，大家搞清楚这个它现在表示的含义是什么啊啊那同样后边你看现在状态已经是五了，那说明什么呢？说明后面又加了一个三了，那这个三是什么三呢？黄三对吧？黄三引起的这个变化说出了一个黄色的五，那蓝三在哪呢？蓝三还在路上呢，对吧？刚刚这个S次任务读进来还在路上，所以大家看现在我们这个并行任务啊，每一个这个子任务，前后发生的任务和这个并行的任务，它当前处理的数据都有可能不一样，对吧，有些数据在路上，有些数据可能是正在处理啊，那而且前后发生的这个数据肯定是前前面我们这个任务。

快一点吧，他们当前处理的是不同的数据，好，接下来就问题就来了啊，啊就要做一个checkpoint里的保存，那由谁来触发呢？Job manager触发一个指令，它会向每一个source任务去发送一条消息，这个消息呢带有一个checkpoint的ID，大家注意啊，这个二并不是并不是数据，或者说我们想的那个时间啊，它这个是当前的checkpoint ID，就是说二号checkpoint现在要开始进行保存了啊，那现在就发给south任务启动了这个检查点的这个保存操作，South任务接收到这个他的信息之后，招标指令之后呢，就会在当前的这个数据处理完成之后，在后边追加一个插入一个。

RI对吧？哎，就是现在就是我们说的就真正的产生进来了，在任务这里面插入到了数据流里面啊，所以接下来我们看到啊，接下来我们这个SS任务啊，这里边我们还是先看这这这个画面啊，这这个图里边，大家会看到后面其实已经有一点变化了啊，为什么呢？这里边还是蓝三黄四对吧，这个时候就接收到了专的指令，但是这个时候专专门这个souths任务，在把这个Barry插进来的过程当中，后边的任务是不是根本没闲着呀。对吧，你看这里边这个当然还是二跟五啊，但是后边你看这个之前我们上面还在路上的那个，呃，这个黄二和下边这个还在路上的蓝二就都已经写入到think系统里面去了，后面的这个任务并没闲着，然后接下来我们这个S任务已经既然已经收到这个呃指令啊，而且也已经插入这个barrier了，那接下来是不是任务首先自己得把自己的这个状态做一个保存啊，哎，所以自己的状态是什么？就是自己这这篇一亮呗，所以蓝三黄四分别把自己的状态做一个保存，大家看，现在就把这状态写入到了远程的这个存储空间里面啊，当然这个point组织的时候，那应该是有对应的关系的，对吧，我这里边肯定有一个这个拓扑结构，当前这就是S任务，这两个任务一个是三，一个是四，三黄四写进来，他做完操作之后呢，就会再给。

反馈一个信息，告诉job manager说报告我这边已经搞定了，OK，没有问题了啊，那manager就就知道了，任务这边已经搞定了，然后接下来呢，SS任务会把自己插入的那个barrier向下游广播出来，大家看到了对吧？这个蓝色的，这里边就不是我们前面给的那个，呃，那个就是赵man就发的那个消息了啊，这里边就是真正的那个Barry了，里面同样还有一个二，这个二表示二号，对吧。

所以因为大家知道，其实本身这个拆有可能非常频繁，有可能前面这在这个处理流程里边，还有前面那个一号还没做完的，完全是有这种可能的，所以这里边我们会把这个直接指定出来啊，这是二号拆point广播出来，然后接下来那就是我们后面这个萨任务接收到这个bar的时候，到底该做什么操作了，对吧？你看他在跟manager自己保存这个状态，然后跟manager确认，然后发出向下有广播这个barrier，在这个过程当中，S任务在做这些事的时候，他是暂停不能去读取下面的数据的，对吧？啊，因为你这个时候状态还没保存好呢，你肯定不能继续去读啊，但是后边的任务根本没有闲着。

你看上面我们这里边萨米一本是二对吧，现在变成四了，之前我们这里边不是在路上的有一个蓝二吗？现在这个蓝二已经叠加进来了，所以输出了一个蓝四对吧？这个项目一本啊，这个偶数求和还在做计算，然后这个奇数求和也没闲着，之前这里面不是有一个蓝三吗？那现在这个蓝三也已经没有了，也已经进来了，所以五加三得到了一个八，输出了一个蓝八。后边的任务并不停着，这就是谁做保存谁暂停，其他的任务完全没影响，对吧，大家各自做各自的啊，你你最后就是每个人到这个位置的时候，下去拍一张照片，最后再P起来就完事了，别人并不影响。啊，那接下来就是我们的这个萨姆任务，要接收到当前的这个，首先我们看上面这个萨even偶数求和啊，它首先接接收到了一个蓝色的这个barrier，大家看就是我主要是看这个萨米伊文，它的这个像像它传输数据的这两个箭头，对吧？首先我接收到的是蓝色的这个barrier，那我们说这个它要直接保存吗？

不能保存啊，不能保存，因为我们现在要保存的这个状态是处理完某个数据之后的所有任务，都处理完某个数据之后的状态，那我们现在应该是处理完哪个数据呢？蓝三黄四嘛，对吧，你现在既然是两个圆，那当然就是这两个数据都都要作为我们当前的一个标准，蓝三黄四，蓝三是处理完了，黄四还在路上呢，明明没处理完的，对吧？所以你必须得等到两个流里边的这个barrier都到齐，你才能够保存自己的这个状态。这就是我们所说的要瑞尔对齐，要等到瑞尔都都到齐了啊，那所以接下来这里边这个呃萨这里啊，收到这个蓝色的barrier之后，我在这里是等着的，并没有直接去做保存操作，哎，那然后这里边就稍微的会有一点麻烦大家看到啊，就是前面这里边来了这个呃蓝色的Barry之后，假如说我这个黄四还来的比较慢，就这里边我们看这个黄色马上就来了，对吧，那假如说这里边这个传输又慢一点，因为这个本来网络传输是没准的嘛，那我是前面这个S任务已经把蓝四都读进来，而且也已经发送到下游来了。

因为大家想这里边这个Barry已经发过来之后，你在这儿等着，但是上游这个数据还在往后传啊，那你后面这个蓝色也已经发过来了，这怎么办呢。有同学可能想，你如果要是这么去做的话，那那个蓝四是直接要叠加，叠加在这个四之上吗？我直接去做操作吗。注意，如果说是这个蓝色的Barry已经来了的话，它后边继续再来的蓝色的数据就必须先缓存起来，起来不能直接做计算，即使我现在空闲着没有没有数据在做计算，也不能直接把它拿来做计算。

就必须等着，为什么呢？因为我要的状态蓝三黄四处理完对吧，你现在蓝三黄四没处理完呢，我也没有保存呢，你不能直接就把后面的数就直接这个状态就叠加进来呀，要不然的话，你这个数本来它应该属于下一个拆框的才能保存它的状态，你现在就保存到这个拆框里的乱了对吧？那所以接下来就是如果蓝四先来，要继续等着，大家看蓝四继续在这等着，但是黄四继续来了的话，这个黄色班主任没来对吧？哎，我等的就是他嘛，蓝三黄四等的就是他，他来了之后我要继续叠加，在之前的四基础上加成八，然后输出一个八，哎，所以在这个过程当中，就是我们所谓的这个算法，主要就是处理这些场景，对吧，到底是怎么样去传输barrier，然后收到barrier的时候，对齐要要等到对齐，那对齐之前接收到的数据又怎么样去做操作，这就是算法的核心所在。

哦，然后之后如果说这个bar瑞尔后面这个黄二也来了之后，那现在我就知道了，对吧，现在Barry已经对齐了，当前我真的可以做这个呃，二号checkpoint的保存了，那接下来呢，哎，我们这里边就可以把当前的状态大家看就是八对吧？八大家知道是这个蓝三黄四嘛，那前面是不是应该是二加二加四得到的八没有问题对吧？呃，然后同样下面这个基数求和呢，也应该同样有一个拆对齐的过程，它加的就是啊，一加一加三加三加。呃，没有五了对吧，所以就是两个一两个三，所以也是八，最后这里边存的状态就是八八，第二个这个萨任务啊，这里边就是八八存进来啊，然后存进这个来之后，接下来就可以继续做计算了，对吧？他做存储的时候，你看前面这个任务也没停，也没闲着呀，这个已经读到六了对吧？这个已经读到八了，这个这个完全没闲着，这个这些都在路上的，他在做存，他在做这个保存米的过程当中，这些数据都已经发过来放在它缓冲区了啊那呃，这里边我把这个保存完了之后，接下来怎么办呢？当然也要通知job manager我完成了对吧？啊那另外这里边我还有一个之前缓存下的没没处理完的这个数据呢，对吧？那接下来我就是先按照顺序先处理当时缓冲起来没处理的这个数据，然后接下来再继续往后读取数据，继续处理就完事了啊，这就是这个整体的一个处理流程那。

当然接下来这个它处理完了之后，这个保存完了之后，这个Barry尔也要朝下游广播，对吧，因为只有下游只有一个任务，那顺序的发送下去就完事了，所以S任务接收到的时候也是发现，诶当前就是蓝三黄色处理完成之后，哦，接下来收到了一个barer，我去做一个自己状态的保存，然后再把它存盘就完事了，等到所有任务都处理完成之后，然后接下来呢，我再去，呃，全部都通知到job manager，那job manager那边就知道了，我现在所有的状态都已经保存成功，接下来诶，那就可以去确认，大家看这个所有任务都通知给draw manager对吧，都已经保存完成了，那draw manager就跟大家确认，说好当前一个checkpoint正式完成，如果接下来你们挂了的话，我们下一步就从当前这个拆point上恢复出来。

就是整个处理的一个流程。检查点的算法对吧？啊，那所以这个过程大家还是要注意一下它的一些细节的。