重磅！ | Flink1.14新特性预览

今天药忘吃喽~ 2022-09-10 13:22 291阅读 0赞

欢迎加博主微信：threeknowbigdata，拉你进大数据群、Flink流计算群

大家好，我是土哥。

目前在某互联网大厂担任大数据算法工程师。

今天在查看Flink源码时，发现`Flink1.14`修复了很多`bug`，并且提了很多`PR`,其中完成了`33`个重要的新特性及优化。

`Bug`修复可见部分截图：

![3cd2c4ca4b041b1860bd1bbcb22238a2.png][]

`Improvement`可见部分截图：

![cd22d2c2a71507f4a5636b778090b616.png][]

新版本预计1-2周之内会发布，下面我将带领大家查看一下 `Flink1.14`的新特性都有哪些？

## 1、流批一体优化 ##

流批一体其实从 `Flink 1.9` 版本开始就受到持续的关注，它作为社区 `RoadMap` 的重要组成部分，是大数据实时化必然的趋势。但是另一方面，`传统离线`的计算需求其实并不会被实时任务完全取代，而是会长期存在。

基于这种背景下，Flink 社区认定了`实时离线一体化的技术路线`是比较重要的技术趋势和方向。

在Flink1.14以前，社区在流批一体方面做了很多的工作。

### 1.1、流批一体的痛点 ###

`重点：` Flink 在引擎层面，API 层面和算子的执行层面上做到了真正的流与批用同一套机制运行。`但是在任务具体的执行模式上会有 2 种不同的模式`。

1.  流模式： 对于源源不断的数据流，统一采用流模式，所有计算节点是通过 `Pipeline` 模式去连接的，Pipeline 是指上游和下游计算任务是同时运行的，随着上游不断产出数据，下游同时在不断消费数据。

![fb238778119e3f2605c83d5d8b19ee28.png][]

1.  批模式：对于有限的数据流，可以把它当成批的执行模式。批的执行模式虽然也有`eventTime`，但是对于 `watermark` 来说`只支持正无穷`。对数据和 state 排序后，它在任务的调度和 shuffle 上会有更多的选择。

流批的执行模式最主要的区别在于：批执行模式`会有落盘的中间过程`，只有当前面任务执行完成，下游的任务才会触发，这个容错机制是通过 shuffle 进行容错的。

两者的区别：

对于流的执行模式来说，它没有落盘的压力，同时容错是基于数据的分段，通过不断对数据进行打点 Checkpoint 去保证断点恢复；

对于批执行模式来说，因为要经过 `shuffle` 落盘，所以对磁盘会有压力。但是因为数据是经过排序的，所以对批来说，后续的计算效率可能会有一定的提升。同时，在执行时候是经过分段去执行任务的，无需同时执行。在容错计算方面是根据 stage 进行容错。

### 1.2、流批一体的优化 ###

针对上述问题，`Flink 1.14` 的优化点主要是针对在流的执行模式下，`如何去处理有限数据集`。

之前处理无限数据集，和现在处理有限数据集最大的区别在于引入了 “任务可能会结束” 的概念。在这种情况下带来一些新的问题，如下图:

![5ff4e86a0492c9b615feef5e4240a1a3.png][]

在流的执行模式下的 Checkpoint 机制

1.  对于无限流，它的 `Checkpoint` 是由所有的 source 节点进行触发的，由 source 节点发送 Checkpoint Barrier ，当 Checkpoint Barrier 流过整个作业时候，同时会存储当前作业所有的 state 状态。
2.  但在有限流的 Checkpoint 机制中，Task 是有可能提早结束的。上游的 Task 有可能先处理完任务提早退出了，但下游的 Task 却还在执行中。在同一个 stage 不同并发下，有可能因为数据量不一致导致部分任务提早完成了。

这种情况下，在后续的执行作业中，如何进行 Checkpoint？

在 1.14 中，`JobManager` 动态根据当前任务的执行情况，去明确 `Checkpoint Barrier` 是从哪里开始触发。同时在部分任务结束后，后续的 Checkpoint 只会保存仍在运行 Task 所对应的 stage，通过这种方式能够让任务执行完成后，还可以继续做 Checkpoint ，在有限流执行中提供更好的容错保障。

![7d258ff4e709e083489f86ad91a83b17.png][]

1.  Task 结束后的两阶段提交协议

我们在部分 `Sink` 使用上，如下图的 Kafka Sink 上，涉及到 Task 需要依靠 Checkpoint 机制，进行二阶段提交，从而保证数据的 Exactly-once 一致性。

![621b280c3977bbc20248e0eeb8c85f5c.png][]

在 Checkpoint 过程中，每个算子只会进行预提交操作。比如数据会提交到外部的临时存储目录下，所有任务都完成这次 Checkpoint 后，会收到一个信号，之后才会执行正式的 commit，把所有分布式的临时文件一次性以事务的方式提交到外部系统。

`这种算法在当前有限流的情况下，作业结束后并不能保证有 Checkpoint，那么最后一部分数据如何提交？`

在 1.14 中，这个问题得到了解决。Task 处理完所有数据之后，必须等待 Checkpoint 完成后才可以正式的退出，这是流批一体方面针对有限流任务结束的一些改进。

## 2、CheckPoint优化 ##

### 2.1、现有checkpoint机制痛点 ###

目前 Flink 触发 `Checkpoint` 是依靠`barrier` 在算子间进行流通，`barrier` 随着算子一直往下游进行发送，当算子下游遇到 `barrier` 的时候就会进行快照操作，然后再把 `barrier` 往下游继续发送。对于多路的情况我们会把 `barrier` 进行对齐，把先到 `barrier` 的这一路数据暂时性的 `block`，等到两路 `barrier` 都到了之后再做快照，最后才会去继续往下发送 `barrier`。

![4d4f69bddf05e5635c5a587406a47656.png][]

现有的 Checkpoint 机制存在以下问题：

1.  反压时无法做出 `Checkpoint` ：在反压时候 `barrier` 无法随着数据往下游流动，`造成反压的时候无法做出 Checkpoint`。但是其实在发生反压情况的时候，我们更加需要去做出对数据的 `Checkpoint`，因为这个时候性能遇到了瓶颈，是更加容易出问题的阶段；
2.  Barrier 对齐阻塞数据处理 ：阻塞对齐对于性能上存在一定的影响；
3.  恢复性能受限于 `Checkpoint`间隔 ：在做恢复的时候，`延迟受到多大的影响很多时候是取决于 Checkpoint 的间隔，间隔越大，需要 replay 的数据就会越多，从而造成中断的影响也就会越大`。但是目前 Checkpoint 间隔受制于持久化操作的时间，所以没办法做的很快。

### 2.2、Flink1.13提出 Unaligned Checkpoint ###

1.  针对Checkpoint痛点，Flink1.14以前就一直在持续优化，（未对齐） `Unaligned Checkpoint` 就是其中一个机制。`barrier`算子在到达 input buffer 最前面的时候，`就会开始触发 Checkpoint 操作。它会立刻把 barrier 传到算子的 OutPut Buffer 的最前面，相当于它会立刻被下游的算子所读取到`。通过这种方式可以使得 barrier 不受到数据阻塞，解决反压时候无法进行 Checkpoint 的问题。
2.  当我们把 `barrier` 发下去后，需要做一个短暂的暂停，暂停的时候会把算子的 `State` 和 `input output buffer` 中的数据进行一个标记，以方便后续随时准备上传。对于多路情况会一直等到另外一路 barrier 到达之前数据，全部进行标注。
3.  通过这种方式做 Checkpoint 的时候，不需要对 barrier 进行对齐，`唯一需要做的停顿就是在整个过程中对所有 buffer 和 state 标注`。这种方式可以很好的解决反压时无法做出 Checkpoint ，和 Barrier 对齐阻塞数据影响性能处理的问题。

![367aa8e62d892847224bc5dc48e9c1f1.png][]

### 2.3、Generalized Incremental Checkpoint ###

Flink 1.14 提出通过`Generalized Incremental Checkpoint`（广义增量检查点）减少 Checkpoint 间隔，

如下图所示，在 `Incremental Checkpoint` 当中，先让算子写入 state 的 changelog。`写完后才把变化真正的数据写入到 StateTable 上`。state 的 changelog 不断向外部进行持久的存储化。在这个过程中我们其实无需等待整个 `StateTable` 去做一个持久化操作，我们只需要保证对应的 `Checkpoint` 这一部分的 `changelog` 能够持久化完成，就可以开始做下一次 `Checkpoint`。StateTable 是以一个周期性的方式，独立的去对外做持续化的一个过程。

![5307c7dec214740a30eb6cdaa7ee5c30.png][]

这两个过程进行拆分后，就有了从之前的需要做`全量持久化 (Per Checkpoint)`变成 `增量持久化 (Per Checkpoint) + 后台周期性全量持久化`，从而达到同样容错的效果。在这个过程中，每一次 Checkpoint 需要做持久化的数据量减少了，从而使得做 Checkpoint 的间隔能够大幅度减少。

其实在 RocksDB 也是能支持 `Incremental Checkpoint` 。但是有两个问题：

1.  `RocksDB 的 Incremental Checkpoint 是依赖它自己本身的一些实现`，当中会存在一些数据压缩，`压缩所消耗的时间以及压缩效果具有不确定性，这个是和数据是相关的`；
2.  `只能针对特定的 StateBackend 来使用`，目前在做的 `Generalized Incremental Checkpoint` 实际上能够保证的是，它与 `StateBackend` 是无关的，从运行时的机制来保证了一个比较稳定、更小的 Checkpoint 间隔。

`Generalized Incremental Checkpoint` 有望在1.14。1 中完成。

[3cd2c4ca4b041b1860bd1bbcb22238a2.png]: /images/20220829/bccd4e2679094271b03171c4584290d1.png
[cd22d2c2a71507f4a5636b778090b616.png]: /images/20220829/ec7e460fd2a74a40ae37ff9972ef1a83.png
[fb238778119e3f2605c83d5d8b19ee28.png]: /images/20220829/030b7b935cf14c6d988b976a39caedff.png
[5ff4e86a0492c9b615feef5e4240a1a3.png]: /images/20220829/009f89ee7ae04784badc6d608725c954.png
[7d258ff4e709e083489f86ad91a83b17.png]: /images/20220829/e19cda8d9e874ccd86e3b11392768c52.png
[621b280c3977bbc20248e0eeb8c85f5c.png]: /images/20220829/10231febc6a1447e9ba093d6de582f49.png
[4d4f69bddf05e5635c5a587406a47656.png]: /images/20220829/24ae7bb771824c54bab728b1347a1866.png
[367aa8e62d892847224bc5dc48e9c1f1.png]: /images/20220829/54371130716a476bbae4e03ec3724f2f.png
[5307c7dec214740a30eb6cdaa7ee5c30.png]: /images/20220829/e328163b088542a594ce2d279b81a598.png