关于Flink:Flink-114-新特性预览

本文由社区志愿者陈政羽整顿，内容源自阿里巴巴技术专家宋辛童 (五藏) 在 8 月 7 日线上 Flink Meetup 分享的《Flink 1.14 新个性预览》。次要内容为：

1. 简介
2. 流批一体
3. Checkpoint 机制
4. 性能与效率
5. Table / SQL / Python API
6. 总结

此文章为 8 月 7 日的分享整顿，1.14 版本最新进展采纳正文的形式在文末进行阐明。

一、简介

1.14 新版本本来布局有 35 个比拟重要的新个性以及优化工作，目前曾经有 26 个工作实现；5 个工作不确定是否能准时实现；另外 4 个个性因为工夫或者自身设计上的起因，会放到后续版本实现。[1]

1.14 绝对于历届版本来说，囊括的优化和新增性能点其实并不算多。其实通过观察发版的节奏能够发现，通常在 1-2 个大版本后都会公布一个变动略微少一点的版本，次要目标是把一些个性稳定下来。

1.14 版本就是这样一个定位，咱们称之为品质改良和保护的版本。这个版本预计 8 月 16 日进行新个性开发，可能在 9 月份可能和大家正式见面，有趣味能够关注以下链接去跟踪性能公布进度。

• Wiki：https://cwiki.apache.org/conf…
• Jira：https://issues.apache.org/jir…

二、流批一体

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

在实时和离线的需要同时存在的状态下，以往的流批独立技术计划存在着一些痛点，比方：

• 须要保护两套零碎，相应的就须要两组开发人员，人力的投入老本很高；
• 另外，两套数据链路解决类似内容带来保护的风险性和冗余；
• 最重要的一点是，如果流批应用的不是同一套数据处理系统，引擎自身差别可能会存在数据口径不统一的问题，从而导致业务数据存在肯定的误差。这种误差对于大数据分析会有比拟大的影响。

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

Flink 在过来的几个版本中，在流批一体方面做了很多的工作。能够认为 Flink 在引擎层面，API 层面和算子的执行层面上做到了真正的流与批用同一套机制运行。然而在工作具体的执行模式上会有 2 种不同的模式：

• 对于有限的数据流，对立采纳了流的执行模式。流的执行模式指的是所有计算节点是通过 Pipeline 模式去连贯的，Pipeline 是指上游和上游计算工作是同时运行的，随着上游一直产出数据，上游同时在一直生产数据。这种全 Pipeline 的执行形式能够：

  • 通过 eventTime 示意数据是什么时候产生的；
  • 通过 watermark 得悉在哪个工夫点，数据曾经达到了；
  • 通过 state 来保护计算中间状态；
  • 通过 Checkpoint 做容错的解决。

  下图是不同的执行模式：

• 对于无限的数据集有 2 种执行模式，咱们能够把它看成一个无限的数据流去做解决，也能够把它看成批的执行模式。批的执行模式尽管也有 eventTime，然而对于 watermark 来说只反对正无穷。对数据和 state 排序后，它在工作的调度和 shuffle 上会有更多的抉择。

  流批的执行模式是有区别的，最次要的就是批的执行模式会有落盘的两头过程，只有当后面工作执行实现，上游的工作才会触发，这个容错机制是通过 shuffle 进行容错的。

  这 2 者也各有各的执行劣势：

  • 对于流的执行模式来说，它没有落盘的压力，同时容错是基于数据的分段，通过一直对数据进行打点 Checkpoint 去保障断点复原；
  • 然而在批处理上，因为要通过 shuffle 落盘，所以对磁盘会有压力。然而因为数据是通过排序的，所以对批来说，后续的计算效率可能会有肯定的晋升。同时，在执行时候是通过分段去执行工作的，无需同时执行。在容错计算方面是依据 stage 进行容错。

  这两种各有优劣，能够依据作业的具体场景来进行抉择。

Flink 1.14 的优化点次要是针对在流的执行模式下，如何去解决无限数据集。之前解决有限数据集，和当初解决无限数据集最大的区别在于引入了 “ 工作可能会完结 ” 的概念。在这种状况下带来一些新的问题，如下图:

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

  • 对于有限流，它的 Checkpoint 是由所有的 source 节点进行触发的，由 source 节点发送 Checkpoint Barrier，当 Checkpoint Barrier 流过整个作业时候，同时会存储以后作业所有的 state 状态。

  • 而在无限流的 Checkpoint 机制中，Task 是有可能提前完结的。上游的 Task 有可能先解决完工作提前退出了，但上游的 Task 却还在执行中。在同一个 stage 不同并发下，有可能因为数据量不统一导致局部工作提前实现了。这种状况下，在后续的执行作业中，如何进行 Checkpoint？

    在 1.14 中，JobManager 动静依据当前任务的执行状况，去明确 Checkpoint Barrier 是从哪里开始触发。同时在局部工作完结后，后续的 Checkpoint 只会保留仍在运行 Task 所对应的 stage，通过这种形式可能让工作执行实现后，还能够持续做 Checkpoint，在无限流执行中提供更好的容错保障。

• Task 完结后的两阶段提交

咱们在局部 Sink 应用上，例如下图的 Kafka Sink 上，波及到 Task 须要依附 Checkpoint 机制，进行二阶段提交，从而保证数据的 Exactly-once 一致性。

具体能够这样说：在 Checkpoint 过程中，每个算子只会进行筹备提交的操作。比方数据会提交到内部的长期存储目录下，所有工作都实现这次 Checkpoint 后会收到一个信号，之后才会执行正式的 commit，把所有分布式的临时文件一次性以事务的形式提交到内部零碎。

这种算法在以后无限流的状况下，作业完结后并不能保障有 Checkpoint，那么最初一部分数据如何提交？

在 1.14 中，这个问题失去了解决。Task 解决完所有数据之后，必须期待 Checkpoint 实现后才能够正式的退出，这是流批一体方面针对无限流工作完结的一些改良。

三、Checkpoint 机制

1. 现有 Checkpoint 机制痛点

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

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

• 反压时无奈做出 Checkpoint：在反压时候 barrier 无奈随着数据往上游流动，造成反压的时候无奈做出 Checkpoint。然而其实在产生反压状况的时候，咱们更加须要去做出对数据的 Checkpoint，因为这个时候性能遇到了瓶颈，是更加容易出问题的阶段；
• Barrier 对齐阻塞数据处理：阻塞对齐对于性能上存在肯定的影响；
• 复原性能受限于 Checkpoint 距离：在做复原的时候，提早受到多大的影响很多时候是取决于 Checkpoint 的距离，距离越大，须要 replay 的数据就会越多，从而造成中断的影响也就会越大。然而目前 Checkpoint 距离受制于长久化操作的工夫，所以没方法做的很快。

2. Unaligned Checkpoint

针对这些痛点，Flink 在最近几个版本始终在继续的优化，Unaligned Checkpoint 就是其中一个机制。barrier 算子在达到 input buffer 最后面的时候，就会开始触发 Checkpoint 操作。它会立即把 barrier 传到算子的 OutPut Buffer 的最后面，相当于它会立即被上游的算子所读取到。通过这种形式能够使得 barrier 不受到数据阻塞，解决反压时候无奈进行 Checkpoint 的问题。

当咱们把 barrier 发下去后，须要做一个短暂的暂停，暂停的时候会把算子的 State 和 input output buffer 中的数据进行一个标记，以不便后续随时筹备上传。对于多路状况会始终等到另外一路 barrier 达到之前数据，全副进行标注。

通过这种形式整个在做 Checkpoint 的时候，也不须要对 barrier 进行对齐，惟一须要做的进展就是在整个过程中对所有 buffer 和 state 标注。这种形式能够很好的解决反压时无奈做出 Checkpoint，和 Barrier 对齐阻塞数据影响性能解决的问题。

3. Generalized Incremental Checkpoint [2]

Generalized Incremental Checkpoint 次要是用于缩小 Checkpoint 距离，如左图 1 所示，在 Incremental Checkpoint 当中，先让算子写入 state 的 changelog。写完后才把变动真正的数据写入到 StateTable 上。state 的 changelog 一直向内部进行长久的存储化。在这个过程中咱们其实无需期待整个 StateTable 去做一个长久化操作，咱们只须要保障对应的 Checkpoint 这一部分的 changelog 可能长久化实现，就能够开始做下一次 Checkpoint。StateTable 是以一个周期性的形式，独立的去对外做继续化的一个过程。

这两个过程进行拆分后，就有了从之前的须要做 全量长久化 (Per Checkpoint) 变成 增量长久化 (Per Checkpoint) + 后盾周期性全量长久化，从而达到同样容错的成果。在这个过程中，每一次 Checkpoint 须要做长久化的数据量缩小了，从而使得做 Checkpoint 的距离可能大幅度缩小。

其实在 RocksDB 也是能反对 Incremental Checkpoint。然而有两个问题：

• 第一个问题是 RocksDB 的 Incremental Checkpoint 是依赖它本人自身的一些实现，当中会存在一些数据压缩，压缩所耗费的工夫以及压缩成果具备不确定性，这个是和数据是相干的；
• 第二个问题是只能针对特定的 StateBackend 来应用，目前在做的 Generalized Incremental Checkpoint 实际上可能保障的是，它与 StateBackend 是无关的，从运行时的机制来保障了一个比较稳定、更小的 Checkpoint 距离。

目前 Unaligned Checkpoint 是在 Flink 1.13 就曾经公布了，在 1.14 版本次要是针对 bug 的修复和补充，针对 Generalized Incremental Checkpoint，目前社区还在做最初的冲刺，比拟有心愿在 1.14 中和大家见面。[2]

四、性能与效率

1. 大规模作业调度的优化

• 构建 Pipeline Region 的性能晋升：所有由 pipline 边所连贯形成的子图。在 Flink 任务调度中须要通过辨认 Pipeline Region 来保障由同一个 Pipline 边所连贯的工作可能同时进行调度。否则有可能上游的工作开始调度，然而上游的工作并没有运行。从而导致上游运行完的数据无奈给上游的节点进行生产，可能会造成死锁的状况
• 工作部署阶段：每个工作都要从哪些上游读取数据，这些信息会生成 Result Partition Deployment Descriptor。

这两个构建过程在之前的版本都有 O (n^2) 的工夫复杂度，次要问题须要对于每个上游节点去遍历每一个上游节点的状况。例如去遍历每一个上游是不是一个 Pipeline 边连贯的关系，或者去遍历它的每一个上游生成对应的 Result Partition 信息。

目前通过引入 group 概念，假如已知上下游 2 个工作的连贯形式是 all-to-all，那相当于把所有 Pipeline Region 信息或者 Result Partition 信息以 Group 的模式进行组合，这样只需晓得上游对应的是上游的哪一个 group，就能够把一个 O (n^2) 的复杂度优化到了 O (n)。咱们用 wordcount 工作做了一下测试，比照优化前后的性能。

从表格中能够看到构建速度具备大幅度晋升，构建 Pipeline Region 的性能从秒级晋升至毫秒级别。工作部署咱们是从第一个工作开始部署到所有工作开始运行的状态，这边只统计了流，因为批须要上游完结后能力完结调度。从整体工夫来看，整个工作初始化，调度以及部署的阶段，大略可能缩小分钟级的工夫耗费。

2. 细粒度资源管理

细粒度资源管理在过来很多的版本都始终在做，在 Flink1.14 终于能够把这一部分 API 凋谢进去在 DataSteam 提供给用户应用了。用户能够在 DataStream 中自定义 SlotSharingGroup 的划分状况，如下图所示的形式去定义 Slot 的资源划分，实现了反对 DataStream API，自定义 SSG 划分形式以及资源配置 TaskManager 动静资源扣减。

对于每一个 Slot 能够通过比拟细粒度的配置，咱们在 Runtime 上会主动依据用户资源配置进行动静的资源切割。

这样做的益处是不会像之前那样有固定资源的 Slot，而是做资源的动静扣减，通过这样的形式心愿可能达到更加精密的资源管理和资源的使用率。

五、Table / SQL / Python API

1. Table API / SQL

Window Table-Valued Function 反对更多算子与窗口类型，能够看如下表格的比照：

从表格中能够看出对于原有的三个窗口类型进行增强，同时新增 Session 窗口类型，目前反对 Aggregate 的操作。

1.1 反对申明式注册 Source/Sink

• Table API 反对应用申明式的形式注册 Source / Sink 性能对齐 SQL DDL；
• 同时反对 FLIP-27 新的 Source 接口；
• new Source 代替旧的 connect() 接口。

1.2 全新代码生成器

解决了大家在生成代码超过 Java 最长代码限度，新的代码生成器会对代码进行拆解，彻底解决代码超长的问题。

1.3 移除 Flink Planner

新版本中，Blink Planner 将成为 Flink Planner 的惟一实现。

2. Python API

在之前的版本中，如果有先后执行的两个 UDF，它的执行过程如下图左方。在 JVM 下面有 Java 的 Operator，先把数据发给 Python 上面的 UDF 去执行，执行后又发回给 Java，而后传送给上游的 Operator，最初再进行一次 Python 的这种跨过程的传输去解决，会导致存在很屡次冗余的数据传输。

在 1.14 版本中，改良如右图，能够把它们连贯在一起，只须要一个来回的 Java 和 Python 进行数据通信，通过缩小传输数据次数就可能达到比拟好的性能上的晋升。

3. 反对 LoopBack 模式

在以往本地执行理论是在 Python 的过程中去运行客户端程序，提交 Java 过程启动一个迷你集群去执行 Java 局部代码。Java 局部代码也会和生产环境局部的一样，去启动一个新的 Python 过程去执行对应的 Python UDF，从图下能够看出新的过程其实在本地调试中是没有必要存在的。

所以反对 lookback 模式后能够让 Java 的 opt 间接把 UDF 运行在之前 Python client 所运行的雷同的过程内，通过这种形式：

• 首先是防止了启动额定过程所带来的开销；
• 最重要的是在本地调试中，咱们能够在同一个过程内可能更好利用一些工具进行 debug，这个是对开发者体验上的一个晋升。

六、总结

本文次要解说了 Flink1.14 的次要新个性介绍。

• 首先介绍了目前社区在批流一体上的工作，通过介绍批流不同的执行模式和 JM 节点工作触发的优化改良更好的去兼容批作业；
• 而后通过剖析现有的 Checkpoint 机制痛点，在新版本中如何改良，以及在大规模作业调度优化和细粒度的资源管理下面如何做到对性能优化；
• 最初介绍了 TableSQL API 和 Pyhton 上相干的性能优化。

欢送持续关注发版的一些最新动静以及咱们在后续的 Release 过程中的一些其余技术分享和专题。

正文

[1] 截至到 8 月 31 日，确定进入新版本的是 33 个，已全副实现。

[2] Generalized Incremental Checkpoint 最终在 1.14 中没有实现。