8月22日，Apache Flink 1.9.0 版本正式发布，这也是阿里内部版本 Blink 合并入 Flink 后的首次版本发布。此次版本更新带来的重大功能包括批处理作业的批式恢复，以及 Table API 和 SQL 的基于 Blink 的新查询引擎（预览版）。同时，这一版本还推出了 State Processor API，这是社区最迫切需求的功能之一，该 API 使用户能够用 Flink DataSet 作业灵活地读写保存点。此外，Flink 1.9 还包括一个重新设计的 WebUI 和新的 Python Table API （预览版）以及与 Apache Hive 生态系统的集成（预览版）。

Apache Flink 的目标是开发一个流式处理系统，以统一和支持多种形式的实时和离线数据处理应用程序以及事件驱动的应用程序。在 1.9 版本中，社区在这方面取得了巨大的进步，将 Flink 的流处理和批处理能力集成在了统一的运行之上。

本文将详细介绍 1.9 版本所有主要的新功能、改进、重要变化、以及未来的发展计划。有关更多详细信息，请查看完整版变更日志。

Flink 1.9 版本的二进制分发和源文件可以通过 Flink 项目的下载页面以及文档页面获得。 Flink 1.9 与之前 1.x 版本的@Public API 是兼容的。

你也可以通过 Flink 邮件列表或 JIRA 与社区分享你的想法。

新功能和改进

细粒度批作业恢复 (FLIP-1)

批作业（DataSet、Table API 和 SQL）从 task 失败中恢复的时间被显著缩短了。在 Flink 1.9 之前，批处理作业中的 task 失败是通过取消所有 task 并重新启动整个作业来恢复的，即作业从头开始，所有进度都会废弃。在 1.9 版本中，Flink 将中间结果保留在网络 shuffle 的边缘，并使用这些数据恢复仅受故障影响的 tasks，即处在同一个 failover region （故障区）的 tasks。故障区是指通过 pipelined 数据交换方式连接的 tasks 集合。因此，作业中 batch-shuffle 的连接定义了故障区的边界。有关更多详细信息，请参见 FLIP-1。

要使用这个新的故障策略，需要确保 flink-conf.yaml 中有 jobmanager.execution.failover-strategy: region 的配置。

注意：Flink 1.9 发布包中默认就已经包含了该配置项，不过当从之前版本升级上来时，如果要复用之前的配置的话，需要手动加上该配置。

除此之外，还需要在 ExecutionConfig 中，将 ExecutionMode 设置成 BATCH，这样批作业才能有多个故障区。

“Region” 的故障策略也能同时提升 “embarrassingly parallel” 类型的流作业恢复速度，也就是没有任何像 keyBy、rebalance 等 shuffle 的作业。当这种作业在恢复时，只有受影响的故障区 task 需要重启。对于其他类型的流作业，故障恢复行为与之前的版本一样。

State Processor API (FLIP-43)

直到 Flink 1.9，从外部访问作业的状态仅局限于：Queryable State（可查询状态）实验性功能。此版本中引入了一种新的强大类库，基于 DataSet 支持读取、写入和修改状态快照。在实践上，这意味着：

Flink 作业的状态可以自主构建，通过读取外部系统的数据（例如外部数据库），转换成 savepoint。
Savepoint 中的状态可以使用任意的 Flink 批处理 API 查询（DataSet、Table、SQL）。例如，分析相关的状态模式或检查状态差异以支持应用程序审核或故障排查。
Savepoint 中的状态 schema 可以离线迁移了，而之前的方案只能在访问状态时进行，是一种在线迁移。
Savepoint 中的无效数据可以被识别出来并纠正。

新的 State Processor API 覆盖了所有类型的快照：savepoint，full checkpoint 和 incremental checkpoint。有关更多详细信息，请参见 FLIP-43。

Stop-with-Savepoint (FLIP-34)

“Cancel-with-savepoint” 是停止、重启、fork 或升级 Flink 作业的一个常用操作。然而，当前的实现并没有保证输出到 exactly-once sink 的外部存储的数据持久化。为了改进停止作业时的端到端语义，Flink 1.9 引入了一种新的 SUSPEND 模式，可以带 savepoint 停止作业，保证了输出数据的一致性。可以使用 Flink CLI 来 suspend 一个作业：

bin/flink stop -p [:targetSavepointDirectory] :jobId

最终作业的状态会在成功时设置成 FINISHED 状态，方便用户区别操作是否失败。

更多详细信息，请参见 FLIP-34。

重构 Flink WebUI

社区讨论了现代化 Flink WebUI 的提案，并决定采用 Angular 的最新稳定版来重构这个组件。从 Angular 1.x 跃升到了 7.x 。重新设计的 UI 是 1.9.0 的默认版本，不过仍保留了切换到旧版 WebUI 的按钮。

注：未来，我们不保证旧版 WebUI 的功能是对齐的，且待新版本稳定后将会完全移除旧版 WebUI。

新 Blink SQL 查询处理器预览

在 Blink 捐赠给 Apache Flink 之后，社区就致力于为 Table API 和 SQL 集成 Blink 的查询优化器和 runtime。第一步，我们将 flink-table 单模块重构成了多个小模块（FLIP-32）。这对于 Java 和 Scala API 模块、优化器以及 runtime 模块来说，有了一个更清晰的分层和定义明确的接口。

不仅如此，我们还扩展了 Blink 的 planner 以实现新的优化器接口，所以现在有两个插件化的查询处理器来执行 Table API 和 SQL：1.9 以前的 Flink 处理器和新的基于 Blink 的处理器。基于 Blink 的查询处理器提供了更好的 SQL 覆盖率（1.9 完整支持 TPC-H，TPC-DS 的支持计划在下一个版本实现）并通过更广泛的查询优化（基于成本的执行计划选择和更多的优化规则）、改进的代码生成机制、和调优过的算子实现来提升批处理查询的性能。除此之外，基于 Blink 的查询处理器还提供了更强大的流处理能力，包括一些社区期待已久的新功能（如维表 Join，TopN，去重）和聚合场景缓解数据倾斜的优化，以及内置更多常用的函数。

注：两个查询处理器之间的语义和功能大部分是一致的，但未完全对齐。具体请查看发布说明。

不过，Blink 的查询处理器尚未完全集成。因此，1.9 之前的 Flink 处理器仍然是 1.9 版本的默认处理器，建议用于生产设置。可以在创建 TableEnvironment 时通过 EnvironmentSettings 配置启用 Blink 处理器。被选择的处理器必须要在正在执行的 Java 进程的类路径中。对于集群设置，默认两个查询处理器都会自动地加载到类路径中。当从 IDE 中运行一个查询时，需要在项目中显式地增加一个处理器的依赖。

Table API / SQL 的其他改进

除了围绕 Blink Planner 令人兴奋的进展外，社区还做了一系列的改进，包括：

为 Table API / SQL 的 Java 用户去除 Scala 依赖（FLIP-32）

作为重构和拆分 flink-table 模块工作的一部分，Flink 1.9 版本为 Java 和 Scala 创建了两个单独的 API 模块。对于 Scala 用户而言变化并不大，不过现在 Java 用户在使用 Table API 和 SQL 时，可以不用再引入一堆 Scala 依赖了。

重构 Table API / SQL 的类型系统（FLIP-37）

Flink 1.9 实现了一个新的数据类型系统，以便从 Table API 中移除对 Flink TypeInformation 的依赖，并提高其对 SQL 标准的遵从性，不过还在进行中，预计将在下一版本完工，并且在 Flink 1.9 中，UDF 尚未移植到新的类型系统上。

Table API 的多列和多行转换（FLIP-29）

Table API 扩展了一组支持多行和多列、输入和输出的转换功能。这些转换能够极大简化处理逻辑的实现，同样的逻辑如果使用关系运算符来实现就会相对麻烦。

重构和统一 Catalog API

我们开发了崭新的 Catalog 接口以统一处理内外部的 catalog 及元数据。这项工作虽然主要是为了 Hive 集成（见下文）而启动的，但同时也全面提升了 Flink 在管理 catalog 元数据的整体便利性。在此之前，通过 Table API 或 SQL 定义的表都无法持久化保存；从 Flink 1.9 起，这些表的元数据可以被持久化到 catalog 中。这意味着用户可以在 Hive Metastore Catalog 中创建 Kafka 表，并在 query 中直接引用该表。

SQL API 中的 DDL 支持（FLINK-10232）

到目前为止，Flink SQL 已经支持 DML 语句（如 SELECT，INSERT）。但是外部表（table source 和 table sink）必须通过 Java/Scala 代码或配置文件的方式注册。1.9 版本中，支持 SQL DDL 语句的方式注册和删除表（CREATE TABLE，DROP TABLE）。不过目前还没有增加流特定的语法扩展来定义时间戳抽取和 watermark 生成策略等。流式的需求也将会在下一版本中完整支持。

Hive 集成预览 (FLINK-10556)

Apache Hive 是 Hadoop 生态圈中广泛用于存储和查询海量结构化数据的系统。Hive 除了是一个查询处理器外，还提供了一个叫做 Metastore 的 catalog 来管理和组织大数据集。查询处理器的一个常见集成点是与 Hive 的 Metastore 集成，以便能够利用 Hive 管理的数据。

最近，社区开始为 Flink Table API 和 SQL 实现一个连接到 Hive Metastore 的外部 catalog。在 Flink 1.9 中，用户能够查询和处理存储在 Hive 中多种格式的数据。 Hive 集成还包括支持在 Flink Table API / SQL 中使用 Hive 的 UDF。有关详细信息，请参见 FLINK-10556。

在以前，Table API / SQL 中定义的表一直是临时的。新的 catalog 连接器允许在 Metastore 中持久化存储那些使用 SQL DDL 语句创建的表（参见上文）。这意味着可以直接连接到 Metastore 并注册一个表，例如，Kafka topic 的表。从现在开始，只要 catalog 连接到 Metastore，就可以查询该表。

请注意 Flink 1.9 中提供的 Hive 支持目前还是实验性的，下一个版本中将稳定这些功能，期待大家的反馈。

新 Python Table API 预览(FLIP-38)

Flink 1.9 还引入了 Python Table API 的首个版本（FLIP-38），Flink 的目标是持续完善对 Python 的支持，而这仅仅标志着开始。该功能围绕着 Table API 设计了很薄的一层 Python API 包装器，基本上将 Python Table API 方法的调用都转换为 Java Table API 调用。在 Flink 1.9 版本中，Python Table API 尚不支持 UDF，只是标准的关系操作。在 Python 中支持 UDF 的功能已规划在未来版本的路线图中。

如果想尝试新的 Python API，则需要手动安装 PyFlink。然后，可以看一看文档中的演练并尝试自己探索。社区目前正在准备一个 pyflink 的 Python 包，该包将可以通过 pip 进行安装。

重要变化

Table API 和 SQL 现在是 Flink 发行版的默认配置的一部分。以前，必须通过将相应的 JAR 文件从 ./opt 移动到 ./lib 来启用 Table API 和 SQL。
为了准备 FLIP-39，机器学习类库（flink-ml）已经被移除了。
旧的 DataSet 和 DataStream Python API 已被删除，建议使用 FLIP-38 中引入的新 Python API。
Flink 可以用 Java 9 编译和运行。请注意，与外部系统（connectors，文件系统，reporters）交互的某些组件可能无法工作，目前相应的项目可能不支持 Java 9。

发布说明

如果你计划升级 Flink 到最新的 1.9.0 版本，请查阅发布说明了解更详细的改动列表以及新特性。

创作场景

Apache Flink 1.9 重磅发布：正式合并阿里内部版本 Blink 重要功能