Apache Flink 1.9重磅发布！首次合并阿里内部版本Blink重要功能-阿里云开发者社区

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 项目的目标是开发一个流式处理系统，以统一和支持多种形式的实时和离线数据处理应用程序以及事件驱动的应用程序。在此版本中，我们在这方面取得了巨大的进步，将 Flink 的流处理和批处理能力集成在了统一的运行时之上。

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

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

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

新功能和改进
- 细粒度批作业恢复 (FLIP-1)
- State Processor API (FLIP-43)
- Stop-with-Savepoint (FLIP-34)
- 重构 Flink WebUI
- 预览新的 Blink SQL 查询处理器
- Table API / SQL 的其他改进
- 预览 Hive 集成 (FLINK-10556)
- 预览新的 Python Table API (FLIP-38)
重要变化
发布日志
贡献者列表

新功能和改进

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

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

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

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

“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 模块工作的一部分，我们为 Java 和 Scala 创建了两个单独的 API 模块。对于 Scala 用户来说，没有什么改变。不过现在 Java 用户在使用 Table API 和 SQL 时，可以不用引入一堆 Scala 依赖了。

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

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

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

Table API 扩展了一组支持多行和多列、输入和输出的转换的功能。这些转换显著简化了处理逻辑的实现，同样的逻辑使用关系运算符来实现是比较麻烦的。

崭新的统一的 Catalog API

Catalog 已有的一些接口被重构和（某些）被替换了，从而统一了内部和外部 catalog 的处理。这项工作主要是为了 Hive 集成（见下文）而启动的，不过也改进了 Flink 在管理 catalog 元数据的整体便利性。

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)

此版本还引入了 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。