重磅！Apache Flink 1.16 发布在即！众多新特性全面解读！

Posted 2023-02-20 zhisheng_blog

Apache Flink 继续快速发展，是 Apache 中最活跃的社区之一。Flink 1.16 有超过 230 名贡献者热情参与，完成了 19 个 FLIP 和 900+ 个问题，为社区带来了许多令人兴奋的功能。

Flink 已经成为流处理的主角和事实标准，流批处理统一的概念正在逐渐获得认可，并在越来越多的公司中成功实施。此前，流批一体的概念更强调统一的API和统一的计算框架。今年，基于此，Flink 提出了 Flink-Streaming Warehouse（Streamhouse）的下一个发展方向，进一步升级了流批融合的范围：真正做到了不仅统一计算还统一存储，从而实现统一实时分析。

在 1.16 中，Flink 社区在批处理和流处理方面完成了许多改进：

• • 对于批处理，已经完成了易用性、稳定性和性能方面的全方位提升。1.16 是 Flink 批处理的里程碑版本，也是迈向成熟的重要一步。

  • • 易用性：SQL Gateway 的引入以及与 Hive Server2 的全面兼容，用户可以非常轻松地提交 Flink SQL 作业和 Hive SQL 作业，也可以轻松接入原有的 Hive 生态。

  • • 功能：引入 Join hint，让 Flink SQL 用户手动指定 join 策略，避免不合理的执行计划。Hive SQL 的兼容性已经达到 94%，用户可以以极低的成本将 Hive 迁移到 Flink。

  • • 稳定性：提出推测执行（speculative execution）机制，减少作业的长尾子任务，提高稳定性。改进HashJoin，引入失败回滚机制，避免join失败。

  • • 性能：引入动态分区修剪以减少扫描 I/O 并改进星型模型查询的连接处理。TPC-DS 基准测试有 30% 的改进。我们可以使用混合 shuffle 模式来提高资源使用率和处理性能。

• • 对于流处理，有许多重大改进：

  • • Changelog State Backend 为用户提供秒级甚至毫秒级的检查点，以显着提高容错体验，同时为事务性 Sink 作业提供更小的端到端延迟体验。

  • • Lookup join 广泛用于流处理。查找速度慢、吞吐量低、更新延迟等问题通过普通缓存机制、异步io和可重试查找来解决。这些功能非常实用，解决了用户经常抱怨的痛点，支持更丰富的场景。

  • • 从 Flink SQL 诞生的第一天起，就出现了一些非确定性的操作，可能会导致不正确的结果或异常，给用户带来极大的困扰。在1.16中，我们花了很多精力解决了大部分问题，未来我们会继续改进。

随着流批融合的进一步细化，以及 Flink 表格存储（0.2 已经发布^[1]）的不断迭代，Flink 社区正在一步步推动 Streaming 仓库从概念走向现实和成熟。

了解流式仓库

准确的说，流式仓库就是把数据仓库做成流式的，让整个仓库中每一层的数据实时流动。目标是通过统一的 API 和计算框架，实现具有端到端实时性能的 Streaming Service。请参阅文章^[2]了解更多详情。

批量处理

Flink 是一个统一的流式批处理引擎，由于我们的长期投入，流处理成为了主角。我们还在努力改进批处理，使其成为出色的计算引擎。这使得流批统一的整体体验更加流畅。

SQL Gateway

各个渠道的反馈表明，SQL Gateway^[3] 是用户非常期待的功能，尤其是批处理用户。这个功能终于在 1.16 完成了（设计见 FLIP-91）。SQL Gateway 是对 SQL Client 的扩展和增强，支持多租户和可插拔的 API 协议（Endpoints），解决了 SQL Client 只能服务单个用户，无法与外部服务或组件集成的问题。目前 SQL Gateway 已支持 REST API 和 HiveServer2 协议，用户可以通过 cURL、Postman、HTTP 客户端等多种编程语言连接到 SQL Gateway，提交流作业、批处理作业甚至 OLAP 作业。对于 HiveServer2 Endpoint，请参阅 Hive 兼容性了解更多详细信息。

Hive Compatibility

为了降低 Hive 迁移到 Flink 的成本，我们在这个版本中引入了 HiveServer2 Endpoint 和 Hive 语法改进：

HiveServer2 Endpoint^[4] 允许用户通过 Hive JDBC/Beeline 与 SQL Gateway 交互，并将 Flink 迁移到 Hive 生态系统（DBeaver、Apache Superset、Apache DolphinScheduler 和 Apache Zeppelin）。当用户连接到 HiveServer2 端点时，SQL Gateway 注册 Hive Catalog，切换到 Hive Dialect，并使用批处理模式执行作业。通过这些步骤，用户可以拥有与 HiveServer2 相同的体验。

Hive 语法^[5]已经是大数据处理的事实标准。Flink 改进了与 Hive 语法的兼容性，并增加了对生产中常用的几种 Hive 语法的支持。Hive 语法兼容性可以帮助用户将现有的 Hive SQL 任务迁移到 Flink，方便熟悉 Hive 语法的用户使用 Hive 语法编写 SQL 查询 Flink 中注册的表。兼容性是使用包含超过 12K SQL 案例的 Hive qtest 套件度量的。到目前为止，对于 Hive 2.3，整个 hive 查询的兼容性已达到 94.1%，如果排除 ACID 查询，兼容性已达到 97.3%。

Flink SQL 的Join提示

join 提示是业界常用的解决方案，通过影响执行计划来改善优化器的缺点。Join 是批处理作业中使用最广泛的算子，Flink 支持多种 join 策略。缺少统计信息或优化器的成本模型不佳会导致连接策略选择错误，从而导致执行缓慢甚至作业失败。通过指定一个连接提示，优化器将尽可能选择用户指定的连接策略。它可以避免优化器的各种缺点，并确保批处理作业的生产可用性。

自适应的Hash Join

对于批处理作业，如果输入数据严重倾斜，哈希连接运算符可能会失败，这对用户来说是非常糟糕的体验。为了解决这个问题，我们引入了自适应 Hash-Join，一旦 Hash-Join 在运行时失败，它可以自动回退到 Sort-Merge-Join。这种机制确保了 Hash-Join 在用户完全不知情的情况下始终能够成功并提高稳定性。

批处理作业的推测执行

Flink 1.16 引入了推测执行，以缓解由有问题的节点导致的批处理作业缓慢。有问题的节点可能有硬件问题、意外 I/O 繁忙或 CPU 负载高。这些问题可能会使托管任务的运行速度比其他节点上的任务慢得多，并影响批处理作业的整体执行时间。

当启用推测执行时，Flink 将继续检测慢任务。一旦检测到慢任务，慢任务所在的节点将被识别为有问题的节点，并通过“黑名单列表”机制（FLIP-224）被阻止。调度程序将为慢速任务创建新尝试并将它们部署到未阻塞的节点，而现有尝试将继续运行。新尝试处理相同的输入数据并产生与原始尝试相同的数据。一旦任何尝试首先完成，它将被视为该任务的唯一完成尝试，并且该任务的其余尝试将被取消。

大多数现有 Source 都可以使用推测执行（FLIP-245）。仅当源使用 SourceEvent 时，它必须实现 SupportsHandleExecutionAttemptSourceEvent 以支持推测执行。Sink 尚不支持推测执行，因此推测执行目前不会在 sink 上发生。

Web UI 和 REST API 也得到了改进（FLIP-249），以显示任务的多个并发尝试和阻塞的任务管理器。

Hybrid Shuffle Mode

我们为批处理执行引入了新的混合shuffle^[6]模式。它结合了阻塞 shuffle 和pipeline式 shuffle（在流模式下）的优点。

• • 与阻塞 shuffle 一样，它不需要上游和下游任务同时运行，这允许执行一个资源很少的作业。

• • 与 pipeline 式 shuffle 一样，它不需要在上游任务完成后执行下游任务，这在给定足够资源的情况下减少了作业的整体执行时间。

• • 它通过提供不同的spill策略来适应在保留较少数据和在失败时重新启动较少任务之间的自定义偏好。

注意：此功能是实验性的，默认情况下未激活。

blocking shuffle的进一步改进

我们在此版本中进一步改进了阻塞 shuffle 的可用性和性能，包括自适应网络缓冲区分配、顺序 IO 优化和结果分区重用，允许多个消费者顶点重用同一个物理结果分区，以减少磁盘 IO 和存储空间。这些优化可以为 10 T 规模的 TPC-DS 测试实现 7% 的整体性能提升。此外，还引入了两种压缩比更高的压缩算法（LZO 和 ZSTD），与默认的 LZ4 压缩算法相比，可以进一步减少存储空间和一些 CPU 开销。

动态分区裁剪

对于批处理作业，分区表在生产环境中的使用比非分区表更广泛。目前 Flink 支持静态分区剪枝，优化器在优化阶段将 WHERE 子句中的分区字段相关过滤条件下推到 Source Connector 中，从而减少不必要的分区扫描 IO。星型模式是最常用的数据集市模式中最简单的一种。我们发现很多用户作业不能使用静态分区修剪，因为分区修剪信息只能在执行时确定，这就需要动态分区修剪技术，在运行时根据其他相关表的数据收集分区修剪信息，从而减少不必要的分区扫描 IO 以查找分区表。动态分区修剪的使用已通过 10 T 数据集 TPC-DS 进行验证，可将性能提高高达 30%。

流处理

在 1.16 中，我们在 Checkpoints、SQL、Connector 等领域做了改进，让流计算能够继续领先。

通用的增量检查点

变更日志状态后端旨在使检查点间隔更短且更可预测，此版本已生产ready，致力于使变更日志状态后端适应现有状态后端并提高变更日志状态后端的可用性：

• • 支持状态迁移

• • 支持本地恢复

• • 引入文件缓存以优化恢复

• • 支持基于检查点的切换

• • 改善changelog状态后端的监控体验

  • • 公开变更日志的指标

  • • 将 changelog 的配置暴露给 webUI

Percentile	End to End Duration	Checkpointed Data Size*	Full Checkpoint Data Size*
50%	311ms / 5s	14.8MB / 3.05GB	24.2GB / 18.5GB
90%	664ms / 6s	23.5MB / 4.52GB	25.2GB / 19.3GB
99%	1s / 7s	36.6MB / 5.19GB	25.6GB / 19.6GB
99.9%	1s / 10s	52.8MB / 6.49GB	25.7GB / 19.8GB

表 1：Changelog Enabled / Changelog Disabled 在值状态下的比较（更多详情请参阅此博客^[7]）

RocksDB 重新缩放改进和重新缩放基准

Rescaling 是基于 Apache Flink 构建的云服务的频繁操作，此版本利用 deleteRange 来优化 Incremental RocksDB 状态后端的 rescaling。deleteRange 用于避免大量的扫描和删除操作，对于需要删除的大量状态的升级，恢复速度几乎可以提高 2~10 倍。

提升状态后端的监控体验和可用性

此版本还改进了状态后端的监控体验和可用性。以前，RocksDB 的日志位于自己的 DB 文件夹中，这使得 RocksDB 的调试不是那么容易。此版本让 RocksDB 的日志默认保留在 Flink 的日志目录中。引入了基于 RocksDB 统计的指标，基于统计的指标保存在数据库级别，例如数据库中的总块缓存命中计数。

支持透支(overdraft)缓冲

引入了透支网络缓冲区^[8]的新概念，以减轻在背压期间不间断阻塞子任务线程的影响，可以通过 taskmanager.network.memory.max-overdraft-buffers-per-gate 开启。从 1.16.0 开始，Flink 子任务默认可以请求超过常规配置数量的 5 个额外（透支）缓冲区。这种变化可能会略微增加 Flink Job 的内存消耗，但会大大减少未对齐检查点的检查点持续时间。如果子任务受到下游子任务的背压，并且子任务需要多个网络缓冲区来完成它当前正在做的事情，那么透支缓冲区就会发挥作用。在文档中^[9]阅读有关此内容的更多信息。

将超时与上游子任务的输出缓冲区中未对齐的检查点屏障对齐

此版本更新了从对齐检查点 (AC) 切换到未对齐检查点 (UC) 的时间。在开启UC的情况下，如果配置了execution.checkpointing.aligned-checkpoint-timeout^[10]，每个checkpoint仍然会以一个AC开始，但是当全局checkpoint持续时间超过aligned-checkpoint-timeout时，如果AC还没有完成，那么检查点将切换为未对齐。

以前，一个子任务的切换需要等待来自上游的所有屏障。如果背压服务，下游子任务可能在checkpointing-timeout^[11]内收不到所有barrier，导致checkpoint失败。

在这个版本中，如果在 execution.checkpointing.aligned-checkpoint-timeout 内无法从输出缓冲区发送屏障到下游任务，Flink 让上游子任务先切换到 UC 向下游发送屏障，从而降低检查点背压期间超时的概率。可以在本文档^[12]中找到更多详细信息。

流处理中的非确定性

用户经常抱怨理解流处理的高成本。其中一个痛点是流处理中的不确定性（通常不直观），这可能会导致错误的结果或错误，从 flink sql 可用的第一天开始就存在很长时间。

对于复杂的流式作业，现在可以在运行之前检测并解决潜在的正确性问题。如果问题不能完全解决，可以通过详细信息提示用户调整 SQL，避免引入非确定性问题。可以在文档中^[13]找到更多详细信息。

增强的Lookup Join

Lookup Join 在流处理中被广泛使用，我们引入了一些改进：

• • 为lookup source缓存和相关指标^[14]添加统一抽象，以加快lookup查询

• • 通过作业配置^[15]或查找提示^[16]引入可配置的异步模式 (ALLOW_UNORDERED)，以显着提高查询吞吐量而不影响正确性。

• • 可重试查找机制[10] 为用户提供了更多工具来解决外部系统中的延迟更新问题。

重试异步 I/O 支持

引入了对用户现有代码透明的异步 I/O ^[17]的内置重试机制，允许灵活地满足用户的重试和异常处理需求。

PyFlink

在 Flink 1.15 中，我们引入了一种新的执行模式“线程”模式，其中用户定义的 Python 函数将通过 JNI 在 JVM 中执行，而不是在单独的 Python 进程中执行。但是，它仅支持 Flink 1.15 中 Table API 和 SQL 中的 Python 标量函数。在这个版本中，我们为它提供了更全面的支持。Python DataStream API 以及 Table API 和 SQL 中的 Python 表函数也支持它。

我们还在继续填补 Python API 中最后几个缺失的特性。在这个版本中，我们对 Python DataStream API 提供了更全面的支持，支持侧输出、广播状态等功能，并最终确定了窗口支持。我们还在 Python DataStream API 中添加了对更多连接器和格式的支持，例如添加了对连接器 elasticsearch、kinesis、pulsar、混合源等以及格式 orc、parquet 等的支持。添加了所有这些功能后，Python API 应该与 Java 和 Scala API 中最显着的功能保持一致，用户应该都能够使用 Python 语言流畅地开发大多数 Flink 作业。

其他

新的 SQL 语法

在 1.16 中，我们扩展了更多的 DDL 语法，可以帮助用户更好地使用 SQL：

• • USING JAR^[18]支持动态加载UDF jar，帮助平台开发者轻松管理UDF。

• • CREATE TABLE AS SELECT (CTAS) 支持用户根据现有表和查询创建新表。

• • ANALYZE TABLE^[19] 支持用户手动生成表统计信息，以便优化器生成更好的执行计划。

用于交互式编程的 DataStream 中的缓存

支持通过 DataStream#cache 缓存转换的结果。缓存的中间结果是在第一次计算中间结果时延迟生成的，以便以后的作业可以重用该结果。如果缓存丢失，它将使用原始转换重新计算。目前仅支持批处理模式。此功能对于 Python 中的 ML 和交互式编程非常有用。

History Server和已完成的工作信息增强

我们在此版本中增强了查看已完成作业信息的体验。

• • JobManager / HistoryServer WebUI 现在提供详细的执行时间指标，包括在每个执行状态下花费的持续时间和运行期间累积的忙/闲/背压时间。

• • JobManager / HistoryServer WebUI 现在提供主要 SubTask 指标的聚合，按 Task 或 TaskManager 分组。

• • JobManager / HistoryServer WebUI 现在提供更多环境信息，包括环境变量、JVM 选项和类路径。

• • HistoryServer 现在支持从外部日志归档服务浏览日志[6]。

Protobuf 格式

Flink 现在支持 Protocol Buffers (Protobuf) 格式。这允许您直接在 Table API 或 SQL 应用程序中使用这种格式。

为 Async Sink 引入可配置的 RateLimitingStrategy

Async Sink 在 1.15 中实现，允许用户轻松实现自己的自定义异步sink。我们现在已经扩展它以支持可配置的 RateLimitingStrategy。这意味着sink实现者现在可以自定义其异步sink在请求失败时的行为方式，具体取决于特定的sink。如果没有指定 RateLimitingStrategy，它将默认为当前默认的 AIMDrateLimitingStrategy。

升级说明

我们的目标是让升级尽可能的顺利，但有些变化需要用户在升级到 Apache Flink 1.16 时对程序的某些部分进行调整。请查看发行说明，了解升级期间要进行的调整和要检查的问题列表。

贡献者列表

Apache Flink 社区要感谢使这个版本成为可能的每一位贡献者：

1996fanrui, Ada Wang, Ada Wong, Ahmed Hamdy, Aitozi, Alexander Fedulov, Alexander Preuß, Alexander Trushev, Andriy Redko, Anton Kalashnikov, Arvid Heise, Ben Augarten, Benchao Li, BiGsuw, Biao Geng, Bobby Richard, Brayno, CPS794, Cheng Pan, Chengkai Yang, Chesnay Schepler, Danny Cranmer, David N Perkins, Dawid Wysakowicz, Dian Fu, DingGeGe, EchoLee5, Etienne Chauchot, Fabian Paul, Ferenc Csaky, Francesco Guardiani, Gabor Somogyi, Gen Luo, Gyula Fora, Haizhou Zhao, Hangxiang Yu, Hao Wang, Hong Teoh, Hongbo Miao, HuangXingBo, Ingo Bürk, Jacky Lau, Jane Chan, Jark Wu, Jay Li, Jia Liu, Jie Wang, Jing Ge, Jing Zhang, Jingsong Lee, Jinhu Wu, Joe Moser, Joey Pereira, JunRuiLee, Juntao Hu, JustDoDT, Kai Chen, Krzysztof Dziolak, Kyle Dong, LeoZhang, Levani Kokhreidze, Lihe Ma, Lijie Wang, Liu Jiangang, LuNing (Lucas) Wang, LuNing Wang, Luning (Lucas) Wang, Luning Wang, Marios Trivyzas, Martijn Visser, MartijnVisser, Mason Chen, Matthias Pohl, Metehan Yıldırım, Michael, Mingde Peng, Mingliang Liu, Mulavar, Nie yingping, Niklas Semmler, Paul Lam, Paul Lin, Paul Zhang, PengYuan, Piotr Nowojski, Qingsheng Ren, Qishang Zhong, Ran Tao, Robert Metzger, Roc Marshal, Roman Boyko, Roman Khachatryan, Ron, Ron Cohen, Ruanshubin, Rudi Kershaw, Rufus Refactor, Ryan Skraba, Sebastian Mattheis, Sergey, Sergey Nuyanzin, Shengkai, Shubham Bansal, SmirAlex, Smirnov Alexander, SteNicholas, Steven van Rossum, Suhan Mao, Tan Yuxin, Tartarus0zm, TennyZhuang, Terry Wang, Thesharing, Thomas Weise, Timo Walther, Tom, Tony Wei, Weijie Guo, Wencong Liu, WencongLiu, Xintong Song, Xuyang, Yangze Guo, Yi Tang, Yu Chen, Yuan Huang, Yubin Li, Yufan Sheng, Yufei Zhang, Yun Gao, Yun Tang, Yuxin Tan, Zakelly, Zhu Zhu, Zichen Liu, Zili Sun, acquachen, bgeng777, billyrrr, bzhao, caoyu, chenlei677, chenzihao, chenzihao5, coderap, cphe, davidliu, dependabot[bot], dkkb, dusukang, empcl, eyys, fanrui, fengjiankun, fengli, fredia, gabor.g.somogyi, godfreyhe, gongzhongqiang, hongli, huangxingbo, huweihua, jayce, jaydonzhou, jiabao.sun, kevin.cyj, kurt, lidefu, lijiewang.wlj, liliwei, lincoln lee, lincoln.lil, littleeleventhwolf, liufangqi, liujia10, liujiangang, liujingmao, liuyongvs, liuzhuang2017, longwang, lovewin99, luoyuxia, mans2singh, maosuhan, mayue.fight, mayuehappy, nieyingping, pengmide, pengmingde, polaris6, pvary, qinjunjerry, realdengziqi, root, shammon, shihong90, shuiqiangchen, slinkydeveloper, snailHumming, snuyanzin, suxinglee, sxnan, trushev, tsreaper, unknown, wangfeifan, wangyang0918, wangzhiwu, wenbingshen, xiangqiao123, xuyang, yangjf2019, yangjunhan, yangsanity, yangxin, ylchou, yuchengxin, yunfengzhou-hub, yuxia Luo, yuzelin, zhangchaoming, zhangjingcun, zhangmang, zhangzhengqi3, zhaoweinan, zhengyunhong.zyh, zhenyu xing, zhouli, zhuanshenbsj1, zhuzhu.zz, zoucao, zp, 周磊, 饶紫轩, 鲍健昕, 愚鲤, 帝国阿三

引用链接

[1] 0.2 已经发布: https://flink.apache.org/news/2022/08/29/release-table-store-0.2.0.html
[2] 文章: https://www.alibabacloud.com/blog/more-than-computing-a-new-era-led-by-the-warehouse-architecture-of-apache-flink_598821
[3] SQL Gateway: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/sql-gateway/overview/
[4] HiveServer2 Endpoint: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/hive-compatibility/hiveserver2/
[5] Hive 语法: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/hive-compatibility/hive-dialect/overview/
[6] 混合shuffle: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/ops/batch/batch_shuffle
[7] 此博客: https://flink.apache.org/2022/05/30/changelog-state-backend.html
[8] 透支网络缓冲区: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/memory/network_mem_tuning/#overdraft-buffers
[9] 在文档中: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/memory/network_mem_tuning/#overdraft-buffers
[10] execution.checkpointing.aligned-checkpoint-timeout: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/ops/state/checkpointing_under_backpressure/#aligned-checkpoint-timeout
[11] checkpointing-timeout: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/config/#execution-checkpointing-timeout
[12] 本文档: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/ops/state/checkpointing_under_backpressure/#aligned-checkpoint-timeout
[13] 文档中: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/concepts/determinism
[14] 相关指标: https://cwiki.apache.org/confluence/display/FLINK/FLIP-221%3A+Abstraction+for+lookup+source+cache+and+metric
[15] 作业配置: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/config/#table-exec-async-lookup-output-mode
[16] 查找提示: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/sql/queries/hints/#lookup
[17] 异步 I/O : https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/datastream/operators/asyncio/#retry-support
[18] USING JAR: https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/sql/create/#create-function
[19] ANALYZE TABLE: https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/dev/table/sql/analyze