036__尚硅谷_Flink理论_Flink状态管理（上）算子状态和键控状态

好，这节课我们来讲flink里边的状态管理，呃，这部分内容主要给大家分这么几部分啊，首先说一下什么是flink里边的状态啊，我们之前一直在说flink是有状态的流失处理这个大数据处理引擎啊，那到底什么是这个所谓flink里边的状态，再确认确定一下它的概念，然后后面呢，给大家分开讲，呃，Flink里边分两种，主要分两种状态，一个叫算子状态，呃，就是这个是一种翻译啊，就是英文就是operator state。那还有一种叫监控状态啊，这个所谓监控也就是大家看kid state对吧，就相当于是做了KBY，做了这个分区之后的这样的一个状态，呃，最后再给大家讲一个所谓状态后端的概念啊好，接下来就首先我们看一看所谓的flink里边的状态到底是什么。那我们其实大家已经知道，呃，Flink讲的是有状态的。

流式处理，那其实我们知道流式处理正常来看的话，它其实可以是有状态的，也可以是无状态。大家想想是不是这样，因为在流式处理的过程当中有一些算子，比如说这些任务就是做什么样的计算呢？我只基于某个独立的数据，某个独立的事件。然后只把这个事件诶稍微改一改，或或者做一些处理，做一些计算，然后直接就输出一个数据，那这种操作是不是根本就不涉及到任何的状态啊。就是真正意义上的来一个处理一个，来一个处理一个对不对，大家想一想，我们做过的map flat map filter，是不是就是类似这样的一些操作啊，来一个处理一个对不对啊，根本不涉及到其他的一些东西，所以这种做法其实就是一个无状态的流失处理，之前其实我们在做的一些这个代码实现里边，比方说上次我们实现有一个代码是超过一定温度的时候，或者说低于一定温度的时候直接报警，大家还记得那个程序吗？那个程序我们处理的时候有没有状态，有没有。

超超过，比方说诶，我们当时好像不是超过，是低于30度的时候就报警对吧。那个我们有状态吗。我们当时实现过两个程序，一个是如果对判断它在一段时间十秒钟之内温度连续上升的话，就报警，对吧，这是一个程序，另外还有一个程序是假如说判断，诶它对这个温度，温度如果要是低于30度的话就报警，那大家想一想这两个程序，这个温度低于30度的这个有没有状态？对大家回忆的话，其实当时我们在那个，那我们当时也是写了一个process function对吧？呃，做了一个测试处理，我把那个报警信息输出，但其实在里边我们根本没有状态对不对？所以大家其实能想到，在这个过程其实很简单，就是来一个判断一个啊，当时我们写的其实是一个呃，Map function对吧？啊，是一个rich的flat map function，并不是process function，所以当时我们的做法其实就是来一个判断一个，然后处理一个输出，输出一个信息对不对，来一个处理一个，根本没有涉及到任何的状。

这是一种情况，而另外一种情况大家就会想到，像我们另外一个例子，你如果要是说是一段时间内温度连续上升的话，那是不是并不是一个数据来了之后，我就能判断它是不是连续上升，对吧？它是不是要跟之前的某个状态要做比较啊，呃，这种情况下，那它就是需要有状态保持的啊，所以大家看到flink里边本身是可以有无状态流失处理和有状态。

这个流失处理两种情况的，当然我们更熟悉的可能还是有状态处理，因为大家发现比方说我们简单的做一个做一个这个求和的计算，或者说我们做一个word count的计算，大家想其实这么简单的一个操作，中间是不是也得有状态保持啊，对吧？而且你要不的话，是不是就得把所有的数据全部存下来，你最后才能知道它所有的和是多少啊？呃，那如果说我们不想这么做的话，其实你中间只要保持一个之前所有数据，诶，它的和是多少，存一个状态是不是就可以做计算了？哎，所以这其实就是大家比较熟悉的有状态流失处理的过程，好，那所以这个状态到底是什么东西呢？整体来讲，我们可以给一个这样的定义。它其实一个状态，它应该是针对一个任务而言的，对不对？每一个任务它在处理的过程当中可能都会有自己的状态，所以它是由一个任务来维护，并且用来计算某个结果的所有数据，那么呃，就他这些所有的数据都属于这个任务的状态。

从图上看，大家其实就是说我这里边如果要是有一个输入的数据，最后我还要得到一个输出的数据，对不对？在得到输出数据的这个过程当中，假如我还依赖其他的一些东西的话，这些东西就全叫做我当前的状态，对不对？这当前这个任务的状态啊，这就是我们本身这个状态的定义。那从我们实际的这个就是整个架构上来看啊，数据存储上来看的话，状态可以认为就是一个本地变量，因为大家看我们想要保持，就是想要快速的访问这个状态的话，是不是肯定要把这个东西放到内存里面去啊，所以大家可以认为这个东西就是一个本地变量，然后它可以被我们任务的业务逻辑直接访问。

然后整个flink会把这个状态给我们做统一的管理。在这个过程当中，它就包括状态的一致性，还有这个故障之后恢复的这个处理，对吧？还有就是状态怎么样去高效存储以及访问啊，这些东西flink都为都给我们直接把这些事情搞定了，我们不需要去考虑这些事情了啊，所以这也是flink比较方便的一点啊，就是他它会把这个状态怎么样呢？就是整个全做序列化，然后以二进制的形式全部存储起来啊，所以在这个过程当中，其实我们不需要去考虑太多的事情啊，他弗林格整个的内存管理机制是非常非常呃，就是比较积极，也是比较优秀的，当然了这种情况就可以方便我们开发人员呢，专注于应用程序的逻辑，而不要管那么多呃乱七八糟的事情，对吧？在flink里边大家会看到啊，就是状态始终是与特定的算子相关联的，就是每一个算子，大家会想到它是代表了一个任务嘛，你要做的一些操作，所以说这个状态始终是跟它相关的。

那那大家会想到flink既然要把所有的状态都管理起来啊，他要做序列化对吧，做各种各样的这些管理，那是不是运行的时候他要能了解到这些状态的话，我们是不是一开始必须先把这些状态注册好啊，哎，所以大家要注意啊，就是我们一开始之前为什么在做这个处理状态的时候，大家想到我们之前在代码里边要写那么一串很奇怪的东西呢？运行值上下文对吧？还要有一个，呃，所谓的script有一个描述器啊，要获取它的那个状态句柄，它其实就是为了这个事情哦，运运行时flink必须得了解我们所有状态的这些信息，它才能把这些东西管理起来，那flink里面到底有什么样的状态呢？总的说来两种类型，一种就是所谓的operator state算子状态，那么顾名思义，这个状态的作用范围呢，就是限定在当前的这个算子任务。

就一个任务，一个状态啊，这样的一个限定范围，与之对应的另外一种是什么呢？是kid state，就是所谓的监控状态啊，这这个翻译大家不要太在意这个名字啊，或者说你你管它叫分区状态，或者叫什么都可以。顾名思义，既然是k state，它应该是基于什么上面的状态啊？对，肯定就是KBY之后K的stream上，哎有的一些任务，然后出现的状态对不对？呃，那他的状态是靠什么，靠什么来分别靠什么来维护的呢。它就是定义出来的那个key对吧，不同的key就会维护自己不同的这个状态，而且不同的key它的状态也是独立去访问的，对不对啊，当前这个K就访问自己的对应的那个状态，不能访问到其他的K上面去啊，所以就是相当于在这个层级上做了一个分割。

啊，具体来看的话，我们还是给大家详细的做一个介绍啊，首先看这个算子状态operator state到底是什么东西呢？啊，大家看，简单来说的话，我们对于一个算子而言，因为有并行度设置嘛，它可能有并行的几个子任务，对不对，所谓的operator state就是每一个。并行的这个子任务整个对应着一个状态。也就是说所有数据到了这一个子任务里边来的所有数据共享这样的一个状态对不对。啊，那它的作用范围就被限定为当前的这个算子任务，所以由同一个并行任务所处理的所有数据，进来的所有数据都能访问同样的状态。

而状态对于同一个任务而言，它是共享的，那对于其他的任务而言是什么样的呢？啊，那就是隔离的对吧？啊，它不能被其他任务访问，而且甚至也不能被相同算子的另外一个子任务访问啊，就是。它的这个力度就是限定在任务上面对不对，只要你是分开的任务，他们就是同一个状态，只要是不同的任务，那就是不同的状态啊，所以这是这个算子状态的一个定义。然后大家看一看这个flink里边算给算子状态提供了什么样的数据结构呢？啊，主要是这样的三种基本的数据结构，最基本的一种就是所谓的list state列表状态，它其实就是把状态表示成了一个列表，对不对啊，就相当于大家可以认为类似于一个数组一样，一个列表直接把它存起来就可以了啊，那大家会想到就是对于这样的一个呃列表状态而言，你在使用的时候，那就其实还是有一些限制的，对吧？啊，那就是一定要是按照列表状态能够调用的那些方法，那些API去对它进行操作。

除了列表状态之外。还有一个一个状态叫联合列表状态，从这个字面意义上大家其实就能看出来，这个其实跟列表状态也是一回事，对不对，对吧，本质也是表示为一个数据列表这样的一组状态，那它跟这个列表状态的区别在。在哪里呢？它其实整体的这个区别就在于发生故障的时候，或者是从这个呃保存点去启动应用程序的时候，或者说发生故障去恢复一个应用程序的时候，你设置了不同的并行度。它这个程序状态怎么样去处理，怎么样去分发，哎这种他们两者之间的这个机制是不一样的。

那简单来说就是列表状态呢，它会根据我们这个并行度的调整，直接把之前的那个状态重新分组，重新分配，而联合列表状态呢，它相当于是会把之前每一个的那个状态相当于是广播到对应的每一个算子里面去。啊，是这样的一个简单的区别啊，大家大概知道就可以了，然后还有一个广播状态，那广播状态这个broadcast state那就更简单了，大家会想到就是它是什么呢。是不是就是把同一个状态广播给所有的这个算子子任务啊，也就是说像我们上边这张图里边，这里边有一个子任务，它有自己维护的一个状态，下边这个也有一个他自己维护的状态，如果要是广播状态的话，那是不是他们就全一样了啊，所以就是说，呃，假如说一个算子有多项任务，那么而且它的每项任务状态又都相同，我们都让他们访问同样的状态的话，那么这种情况就比较适合用这个广播状态啊，这就看大家具体的这个在应用时候的这个需求啊，实际在应用的过程当中可以跟大家说，其实。

这些状态用的都不多，那用的更多的是什么呢？因为像这种算子状态，它又有另外一个名字叫叫non kid。State就是没有经过用一个键来控制的这这样的一个状态，对不对？没有经过通过K去做分区之后的这个状态，所以更常见的处理，我们要处理的是什么状态呢？对，当然是经过KBY之后的k stream上面定义出来的状态，所以接下来给大家再讲一下这个监控状态。呃，那那大会想这个监控状态它到底是怎么一个情形呢？跟上面我们讲的那个operator state相比，它现在维护和访问的这个力度，就相当于是不是按照每一个子任务一个task来去维护和访问了，而是按照什么呢？按照一个一个key。

所以大家看，对于我们这样一个大家看啊，这这这就类似于K败之后又做了一个聚合操作，对吧，大家可以把这个图简单的理解成这样，如果要是说已经KBY之后再做一个操作的话，他们这里的状态会。由什么来维护呢？大家看这就不是当前这是我们一个task manager，它的那个内存存储空间对不对啊，或者大家认为这是一个slot对吧，它单独自己的一个内存空间，是所有数据都访问同样的一份状态吗？不是，大家看是不是对，就是一个K会维护自己的一个状态啊，哎，这里边粉色的数据来了，你k buy按这个颜色，如果已经做了这个分分组分区的话，粉色的数据来了之后，他只访问自己这个粉色的数状态，蓝色数据来了之后，只访问自己蓝色的这个状态。

互不影响对不对？哎，所以大家看到为什么我们大多数的这个处理都要对这个data stream做KBY呢。是不是在这个意义上，他其实就是做了分区啊，对吧？哎，所以大家看这里边他的状态是自己真的是单独去维护，互不影响，所以即使我slot本身是共享的，大家共享的是一块这个内存空间，但事实上呢。对，他们互相之间的这个state还是不同的对象，对吧，相当于互相之间还是隔离开的啊，这是这个呃，Kid state的一个定义，所以flink是为每一个K都会维护一个状态实力，并且呢，相同的K。

它的所有的数据都会分区到同一个所谓的这个算子任务里面去，那么这个任务就会维护和处理这个K对应所有的这个状态。那那大家看这个KBY，这就相当于是给我们做了数据的一个分区的工作啊，那当然了，当任务处理每一条数据的时候，它的状态的访问范围自动就限定为当前的K，不会跟别的K走串。啊，这是所谓的这个监控状态啊，好，那大家看一看这个监控状态里边我们又能具体处理什么样的数据类型呢？数据结构呢？哎，这个大家看就更好理解一些了，对吧？呃，更。符合我们之前接触过的这个数据类型啊，首先可以有什么样的状态呢？可以有直状态value state啊，这个很简单，Value state就是一个单个的值，对不对啊，之前我们做的那个process function编程的时候，其实就是把对上一个那个温度值，一个double类型的数据是不是就保存成了一个值状态啊，Value state对吧？啊，所以这就是最简单的方式，你要存一个单一的值，就保存成一个value就可以了。

那大家可能会想到，那有时候假如说我要是不仅仅只存之前的一个数据呢。假如我要存之前的一组数据呢，好几个数据呢，哎，对，大家就会想到这种时候我是不是还是得把它保存成一个列表啊，哎，所以这个时候就还可以存成什么呢？哎，把状态可以存成一个列表状态，呃，这个跟那个operator state里边的那个列表状态类似，对不对，大家看到都是一个表示成一组数据列表的这样的一个状态，那直状态和列表状态在。获取它的值和往里边去塞值对吧，去给他更新值操作是不一样的，像这个值状态怎么去获取它的值呢。

大家还记得值状态get操作的时候，是不是直接点value就拿出它的值了？哎，那set操作怎么做呢？对，是直接update对吧，然后把这个值传进去就可以了，那这个list列表状态又是怎么样去操作呢？他的get操作就是直接点get。但是点get它返回的是一个啊，大家会想到肯定是应该是一个列表对吧？啊，但它返回的不是一个list，它返回的是一个ter类型，就可迭代类型啊，是这样的一个数据结构，所以大家如果要调用的时候，还要注意一下它这个不同的这种返回数据类型，然后有不同的这种调用方式，对吧？呃，那当然了，List state如果要去set操作的话，可以直接update。Update的时候，你得把整个一个list子都传进去给它，把整个list update掉。

那大家会想到，那有时候我这个一个一个这个列表操作，我并不想一下子把整个数组这个列表改变掉啊，我有可能是不是只是往那个列表里面追加一个元素啊。呃，大家想到是不是应该有这样的操作啊，所以当然它也有这样的操作，就是有一个点ADDADD的一个元素，就是把我们一个元素是不是追加到列表里面去啊，这个大家就是大概的了解一下，知道它怎么用就可以了，这些方法在文档里面都有啊，大家可以去看文档，除了值状态和列表状态之外呢，还有map state，那对map state这个大家很好理解，那就是状态就表示成了一个k value，对对吧？呃，那这个大家可以认为就是类似于一个哈奇map一样，那么它的get set方法也比较简单，那get的时候就是点get。

传一个K，然后是不是就能拿到它那个value啊啊，这这就是这个get方法，那如果要是set呢，Set的时候是点put啊，大家注意啊，这个不同的状态，它的这个get方法，Get set方法都不一样，这个map state呢，它做这个这个就是set操作的时候是点put put一个一个key value对吧？是这样的一个操作，呃，当然除了这些操作之外，这个map还可以，比方说这个点remove，就是remove，然后传一个K，就相当于把我们这个单独的k value这这一个值给删掉了，对吧？啊，另外还可以就是contains，就表示是否存在这样的一个key对应的值啊，这都是一些常见的操作啊。除了这些大家比较熟悉的数据类型之外，Keep set里边还提供了一个就是相对来讲啊，大家可能觉得有点特殊的数据结构，就是所谓的聚合状态，哎，那什么叫聚合状态呢？聚合状态里面又有两种，一个是这个reducing state，一个是aggregating。

顾名思义，大家一看这个这个名名字啊，就知道这个状态肯定就是可以用来做聚合操作的，对不对啊，它其实这个其实很简单，这个状态整体来讲啊，它可以认为就是一个list set，所以这个list set其实是。本身这个flink做状态管理的一个比较基本比较底层的一个数据结构啊，因为他这个做序列化什么。各方面比较比较容易对吧，比较统一，比较有规范啊，那这个聚合状态跟之前的列表状态又有什么区别呢？这里大家要注意就是呃，列表状态，当时我们点ADD的时候，不是直接把一个状态，就是把一个新的数据要加到那个列表里面去吗。

这里大家注意啊，聚合状态里边也有点儿爱的操作。点的时候不是把这个数据添加到列表里边，而是。直接就聚合到我们之前的那个结果里面去了。啊，所以大家想这个点A是不是就有点像我们直接sum一样，对吧，Sum是不是就是一个简单的聚合对吧？啊，所以它相当于就是直接把这个叠加进去了，直接做聚合了，那这两个reducing和aggregating又有什么区别呢？呃，它俩主要的区别其实就在于就是reducing，大家知道reddu那个操作，那个算子，呃里边相当于应用一个reduce方式，它输出的结果数据类型是不能变的。对吧？啊，就是还必须跟之前输入的那个数据类型一样，而aggregating state呢？呃，它的这个相当于应用这个一般化的聚合函数aggregate方式，它可以得到一个。啊，就是数据结构完全数据类型完全不一样的一个输出结果啊，所以就相当于这个更一般化一些，这个大家就是到时候我们用到的时候再给大家做这个详细的讲解就可以了，大家先有一些概念啊。

好，刚才说了那么多，大家看看这个代码里边到底怎么来实现，哎，这个其实之前我们在代码中已经有过接触了，大家再来复习一下，是要想使用这样一个状态的话，是不是首先得把它声明出来啊，啊，我们常见的一种声明的方法是啊，大家看这个啊，当然这前面加了这个laz lazy这样一个关键字，做了一个懒加载，对吧？或者你也可以比方说在这个open生命周期里边去直接把它生明出来，对不对，那这样也是可以的，大家看声明的过程当中是怎么样呢？首先它的类型，这样一个状态类型是什么呢？比方说声明一个直状态，那这里边是不是就必须它的类型是一个value state呀？然后它里还有这个泛型，这个泛型是它里边我们真正要存的那个数据类型，对不对，比方说一个double类型，这里就是double，它是一个double类型的value state。

那么那后面在定义的过程当中需要怎么样？需要去大家看，需要去用这个运行时上下文去获取状态句柄，对吧？调用get runtime context的get set方法。那么这个获取状态句柄，句柄的时候里边要传一个参数，这个参数是一个状态的描述器啊，这就是说flink管理状态的时候，你得告诉他我们这个状态到底是什么对吧？到底这个状态名字是什么，类型是什么，然后他就可以把它管理起来，给我们做这个序列化，哎，就是不用我们操心了，什么事情都他来管了，所以大家看这里边我们要传的是不是就是一个是这个一个字符串它的名字，另外一个就是啊，对，Class，这是它的那个本身数据类型对吧？啊，把这个本身的这个类结构要给它传进去。

啊，当然这里面我们用的是这个class，也可以直接type对吧？啊，这个都是可以的。那如果读取这个value，读取这个值状态的时候，直接点value，如果要是对它赋值的时候，是不是直接点update啊啊，这个大家针对不同的状态看文档，看它怎么样去操作就可以了。