00:01
了解了检查点,接下来我们再来介绍另外一个flink状态管理机制当中的非常重要的概念,那就是,也就是状态后端。其实对于检查点来说,我们都知道了,他就是。当前flink应用程序里所有状态的一个快照,一个持久化的保存,那本身这些状态它应该是放在哪里的呢?对于一个flink程序而言,所有的算子任务都应该分布式的执行在slot上,而slot是存在于task manager上。呃,本来是task manager上资源的一个划分,划分出了不同的slot。所以本身所有的状态,它应该都是task manager内存里边的一部分。那所以接下来如果说我们要去做快照的话,肯定就会涉及到task manager相关的一些操作。
01:03
而另外呢,我们当前是一个分布式的系统,所有的任务都是并行处理的。那么这就涉及到了所有的task manager,不同的slot,那在这个过程当中,我怎么样能控制他们在同一个时间点去做一个状态的快照持久化的保存呢?这当然就又涉及到了统一的协调和管理。那由谁来管理?自然就是job manager,所以检查点的保存是离不开job manager和task manager他们之间的协调工作的。当然了,这里边既然是涉及到了向外部的持久化入,那就还应该有外部系统的一个协调工作。在。进行检查点保存的时候。首先,那是需要由job manager向所有的task manager发出触发的命令,也就是说,现在我要开始制作检查点了,所有的task manager接收到当前命令的时候,就把当前正在进行的任务。
02:13
所有的状态做一个快照保存。那这个快照保存保存完成之后,当然我们知道本身是在内存里边的一个Java对象,那接下来呢,当然是要做一个做一个序列化,然后把它写入到远程的持久化存储的介质里面去。我们知道一般是。分布式的文件系统,当然如果说我们想把它写到一些特定的数据库也是可以的,那完成之后怎么样才能叫做当前的检查点全部保存成功了呢?并不是说某一个task manager它的状态保存完毕就算成功了。我们当前是一个分布式的系统,如果要能从故障当中恢复出来的话,很显然必须所有的task manager都完成自己当前所有状态的保存。
03:03
所以最终的要求当前检查点真正保存完毕的标志,它是所有task manager都向job manager报告,确认当前检查点完成,这个时候job manager来发发出消息说当前检查点真正的完成。所以我们发现这个过程是分布式进行的。整个这个过程当中,其实是需要很多组件做一系列的协调和控制管理啊,所以整个整个系统来看的话,我们其实最好是有一个专职人员来做这样的一套事情啊,那这个所谓的专职人员就是我们所说的。状态后端。我们这里给出状态后端的具体的定义,那就是在flink当中状态的存储、访问和维护。这就包括了我们本身状态的序列化,序列化持久化保存,发生故障之后的恢复,以及呃,我们说这个病情发生调整之后状态的。
04:11
重组分配,所有的这些都是由一个可插拔的组件决定的,这个组件就叫做state状态后。从定义我们可以看得出来,状态后端其实主要是要负责两方面的事情的,一件事情是本地的状态管理。前面我们说到的。我们直直接从运行上下文里面把它获取出来,定义出来。获取到它的控制句柄,然后直接调用API使用就可以了,底层的那些操作机制我们根本都不需要考考虑,那到底是谁来负责的呢?啊,我们说是flink的底层,Flink底层具体是谁?其实就是状态后端来做这些操作啊,所以它其实是涉及到了我们当前状态,本地状态的一些管理操作的。
05:02
另外一个呢,当然就是需要在进行检查点保存的时候,把本地的状态写入到检查点中,然后存放到远程的持久化存储介质中。这就是状态后端它的功能和用途。那接下来我们再来了解一下状态后端有哪些类型,我们发现了状态后端它是一个可插拔的组件,所以它其实是开箱即用的,一般情况下我们可以进行独立配置,跟业务逻辑是完全没有关系的。Flink里边呢,给我们现在提供的是两类,主要是两类不同的状态后端,一类叫做哈希表状态后端啊,我们发现这其实在源码里边是一个现成的类,叫做。另外一类呢,是内嵌rocks DB状态后端,或者叫嵌入式rocks DB状态后端embedded rocks DB啊,那如果没有特别配置的话,系统里边默认的状态后端是第一种哈希表状态后。
06:15
那这两种状态后端又有哪些特点,他们有哪些区别呢?接下来我们分别做一个介绍,首先看默认的哈希表状态后端,那这种状态后端其实就是我们之前认为的弗对于状态的那种处理方式,就是所有的状态是放在帕manager的内存里面啊,它具体来看的话,那就是说所有的状态就是一个Java对象,一个object。然后所有既然它是抓对象啊,那就是保存在了task manager的GM堆内存上,那所有普通的状态,还有窗口当中收集到的数据,还有触发器trigger啊,所有的这些东西都是状态,他们的存放方式是以键值对的方式形式进行存储,所以我们发现既然是舰职队,那它本质上可以认为就是一个哈希表啊。
07:10
啊,所以它的底层存储结构就是一个哈希表,这也就是为什么这种状态后端叫做哈希表状态后端,它是用这种数据结构进行内部状态的管理和存储的。而对于检查点呢,啊,检查点的话,一般就是我们既然已经当前的这个状态,都是要满足flink里边能够识别,能够管理的那些。类数据类型啊,那这样的话我们就可以做序列化,序列化之后就可以进行持久化的存放了。对于哈希表状态后端,它的检查点一般都是存放在持久化的分布式文件系统上,那最常见的当然就是HDS了。另外也可以通过配置一个叫做checkpoint storage,就是所谓的检查点这样一个选项来单独的进行指定,这些都是可以去做的。
08:09
对于哈希表状态后端来讲,它是把本地状态全部放到内存里面,所以我们就发现了它的优势就在于可以获得最快的读写速度。内存计算就是快嘛啊,所以它的性能是最佳的,那代价呢,当然就是比较耗内存了,当我们当前的状态越来越多,或者说数据量越来越大,窗口数据量非常大的时候,那有可能就会造成我们当前的内存不够用啊。那在。出现这种情况的时候,那得怎么办呢?那就要考虑用其他的状态后端来进行替代了。这就是我们所说的第二种状态,后盾。Embedded DB,一般情况就是把它叫做rock DB状态后啊,那DB其实我们知道它是由Facebook开发出来的,呃,一个开源的,基于k value这样的一个。
09:06
嵌入式的数据库,或者说是存储介质,它是基于Google开发的那个DB的原理其实差不多,那它的特点呢,其实就是可以把当前的数据持久化到本地硬盘里面啊,那当然它也是通过这个k value这种结构进行组织存储的。那如果我们配置了DB状态后端的话,那么所有的状态数据就都会放到DB数据库里。我们可以认为。一个嵌入式的数据库。所以。接下来我们当前所有的状态就变成都放到DB数据库,或者说直接放到本地硬盘里了啊,那默认的话是在task manager的一个本地数据目录里面。我们可以在这个文件本地文件系统里边找到对应的目录,对应的位置。
10:01
那跟前面讲到的哈希表状态后端。所不同的一点,那就是当前的状态。根本就不是Java对内存里,JM对内存里边的一个对象了,现在的话就相当于全放在rocks DB里面了,当然了,Rox DB会有一部分放在内存里进行做缓存的部分啊,但是呢,更大量的数据就会放到本地硬盘当中进行持久化保存。所以这样它的一个特点就是说,我们的读写性能就会差一点,因为你要访问硬盘嘛,既然要访问硬盘,那读写的时候就会涉及到序列化反序列化的过程啊。那呃,如果说我们还要做这个K的比较的话,那它需要按照字节去进行比较,而不是直接调用这个哈,Code或者是ES这样的方法,所以它也会比较慢一点,性能会比较差,那它的好处是什么呢?好处当然非常明显,那就是能够存放的状态更多了,既然它是放到硬盘,自然存放的状态可以更大。
11:06
那对于检查点来讲呢,同样我们也是可以单独配置的,一般情况同样也是写入到远程的持久化文件系统当中,因因为这个的话,我们就不存在性能上的要求嘛,只要能够大量的存放就可以了。另外需要注意的是,Rock DC它执行的是异步快照,也就是说如果说我们配置了DB状态后端的话,我们自然会想到如果说我们是要等到当前的状态。每一个任务啊,如果接收到照manager那边发来的指令,要保存当前状态,做持久化保存了,做快照了,如果说这个时候我就一直等着,等到当前快照做完之后再去处理接下来的数据的话。那很明显,现在因为我要访问稳定硬盘嘛,访问DB这个代价比较大,诶那显然就会造成我们的实时性的降低,延迟就会增加,所以rocks DB它其实是异步执行的,也就是说我们这里边做快照,诶那是那是一个单独的事情。
12:15
想做什么发生这个快照指收到接收到快照指令的时候,我现在就直接开始做当前的。状态的保存,而接下来的数据呢,继续处理,直接读到当前任务里边来同时处理,所以它俩是并行的,这样的话就不会因为我们保存到本地硬盘做快照的这个过程。阻止了数据的处理,不会造成更大的延迟啊,那另外呢,DB还提供了一个,因为我们要保存的。数据量可能比较大,状态比较多,所以D提供了一个增量式的。保存检查点的机制。所谓的增量式,那就是我可以之前先保存一份状态。
13:04
然后我们隔一段时间保存,再保存一个快照,那下一个快照呢,很可能大部分数据跟上一个快照是类似的,所以我不需要保存全量数据,只把它们之间变化的那一部分,相当于是。那个德尔塔保存下来就可以了啊,那这样的话,每一份快照要保存的数据就会更少,所以在很多情况下,这样我们可以大大的提升保存效率。当然了,这样的话,如果我们想要恢复,那就应该要把之前所有的这些快照都要拿过来,然后才能恢复出我们最后的状态。啊,那所以这个过程当中我们就会发现啊,状态后端有两种基本的选择,我们在实际应用当中,到底在什么场景下选择什么样的状态后端呢。通过前面的介绍,他们的对比已经非常的明显了,他们之间最大的区别其实就是在于本地状态放在哪里,一个是放在内存,另外一个是放在D啊,那所以放在内存里面的话,当然就是非常啊,但是它的大小受到内存的限制,如果说我们当前的这个应用是。
14:19
状态不停的增长的话,那么最后内存就会被耗尽,我们就只好停下当前应用,然后再去扩展集群资源了啊。那当然了,就是即使我们能够不停的动态扩展集群资源,它也会带来一个代价,就是我们的成本比较高,因为这个机器内存的这种扩展显然是要耗费大量。大量成本的,所以另外一个选择,那就是rocks DB rocks DB的话,它是硬盘存储,所以它的扩展的代价也很低,所以非常适用于超级海量状态的存储。不过它带来的缺陷就在于当前的读写速度状态的读写速度变慢了,因为每一次访问硬盘要去进行序列化和反序列化啊,那所以整体来讲的话,DB状态后端要比哈希表状态后端慢一个数量级。
15:14
所以在实际应用当中,其实就是一个权衡取舍的过程,我我们就是要看到底是要快还是要要存储,用更低的成本存储更大量更海量的状态呢?啊在实际应用的当中,一般推荐的就是如果状态比较小,而且状态比较稳定,不会随着时间不停的增长,而且我们对于性能要求比较高的时候,那当然是用哈希表状态后端比较合适,而如果说我们的状态比较大,超级海量,或者说是随着时间的推移会不停的增长,增长到海量。或者说是我们对于成本非常的敏感,而对于运行的速度不那么的强调,那这个时候就用DB状态后会比较合适。
16:04
了解了状态后端的不同类型,那接下来我们再来说一下状态后端的配置啊,在状态后端我们知道默认情况下啊,其实就都已经使用的是哈希表状态后端了,在我们不做任何配置的时候,应用程序使用的默认状态后端那是由集群的配置文件flink-com.ya这个文件配置的,那配置的键的名称K名称就是state.back end。啊,那我们知道这一个配置默认配置项啊,是对集群上运行的所有作业,所有应用都是有效的,那我们可以通过更改这一个配置项来去更改默认的状态后端啊,那另外还有一种方法是可以在代码里边去单独设置代码里边设置好的那个状态后端,那就是只针对当前应用有效了,它会覆盖掉默认集群里边的那个配置配配置的默认值。
17:02
啊,那首先我们看一下在集群的配置文件里边怎么样去配置状态后端啊,其实非常简单,就是的这个字段,然后冒号后边跟着的就是指默认情况下呢,就是哈希map,我们可以这样去指定,也可以不指定,那如果说我们想要去配置一个embeded rocks DB state又该怎么配呢?啊,那就把这个直接改成rocks DB就可以了啊,就是小写的rocks DB。这就是非常简单的一个配置的过程,那另外我们说还可以配置当前存放检查点的。路径啊,那这个路径的话就叫做state checkpoint Di啊,那后边的话,一般情况我们用的是一个分布式文件系统的路径,最常见的当然就是HDFS了,一般情况就是这样去做一个配置就可以了。那另外呢,对于这个默认的状态后端来说,我们除了哈希map和rock db2个选项之外,我们发现在这个源码里边。
18:08
这两种不同的状态后端其实就是两个不同的类嘛,所以我们其实它本身是可可插拔的一个组件,我们也可以自定义,那如果要自定义的话,那就必须是实现了状态后端工厂类的一个完全的类的类名,限定的类名就是state back factory,实现了这个接口的类,我们就可以放在这里作为状态后端配置。那除了集群默认的配置文件里面的定义之外,我们还可以为每一个作业或者说每一个应用单独去配置一下当前的状态后端,那这个就需要在代码里边去做配置了,那代码里边怎么配呢?同样也是env在环境里边去做一个简单的配置啊,我们可以在代码里边同样来看一下。代码里面如果要是。
19:01
默认情况下当然也是是哈希表的状态后端了啊,如果我们想单独做配置的话,那就是set state back,这里边我们可以去你一个哈希表,哈希map,这个是完全没有问题的,当然了,这个本身没有必要。另外,我们也可以直接创建一个rocks DB的。Inbed。Rocks DB。State back end。其实我们已经发现了,在这个敲的过程当中,会发现并没有对应的类能够让我们直接使用啊,这要说明的一点是,在idea里边,在我们这个集成开发环境里边,是本身引入的flink相关的包,是不包含rock DB后端的支的,所以我们还需要引入相关的,这个的话其实也是官方给我们提供的,直接就叫做flink rock DB啊,后面这个是的版本,然后下边对应的版本是当前flink的版本,我们可以直接把它copy过来。
20:13
直接放到home键下面。我们把它引入之后。接下来。就有了对应的这个。我们也可以点进去看到它确实是实现了。一个这样一个象类,然后一步的话,我们会看到它本身是实现了这样一个。我们如果是在idea里边进行开发的话,那是必须要引入相关的依赖,而事实上呢,在flink的发行版里边本身是包含了D这样一个依赖的,所以如果说我们直接把当前的代码打包提交到集群环境里边去运行的话,那是没有必要把我们刚刚引入的这个依赖打包进去的啊。
21:07
它是直接可以正常运行的,这就是关于状态后端在代码里边的配置。
我来说两句