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

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

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

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

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

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

新功能和改进

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

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

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

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

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

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

StateProcessorAPI(FLIP-43)

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

Flink作业的状态可以自主构建，通过读取外部系统的数据(例如外部数据库)，转换成savepoint。

Savepoint中的状态可以使用任意的Flink批处理API查询(DataSet、Table、SQL)。例如，分析相关的状态模式或检查状态差异以支持应用程序审核或故障排查。

Savepoint中的状态schema可以离线迁移了，而之前的方案只能在访问状态时进行，是一种在线迁移。

Savepoint中的无效数据可以被识别出来并纠正。

新的StateProcessorAPI覆盖了所有类型的快照：savepoint，fullcheckpoint和incrementalcheckpoint。有关更多详细信息，请参见FLIP-43。

Stop-with-Savepoint(FLIP-34)

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

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

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

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

重构FlinkWebUI

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

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

新BlinkSQL查询处理器预览

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

不仅如此，我们还扩展了Blink的planner以实现新的优化器接口，所以现在有两个插件化的查询处理器来执行TableAPI和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中运行一个查询时，需要在项目中显式地增加一个处理器的依赖。

TableAPI/SQL的其他改进

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

♦为TableAPI/SQL的Java用户去除Scala依赖(FLIP-32)

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

♦重构TableAPI/SQL的类型系统(FLIP-37)

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

♦TableAPI的多列和多行转换(FLIP-29)

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

♦重构和统一CatalogAPI

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

♦SQLAPI中的DDL支持(FLINK-10232)

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

Hive集成预览(FLINK-10556)

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

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

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

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

新PythonTableAPI预览(FLIP-38)

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

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

重要变化

TableAPI和SQL现在是Flink发行版的默认配置的一部分。以前，必须通过将相应的JAR文件从./opt移动到./lib来启用TableAPI和SQL。

为了准备FLIP-39，机器学习类库(flink-ml)已经被移除了。

旧的DataSet和DataStreamPythonAPI已被删除，建议使用FLIP-38中引入的新PythonAPI。

Flink可以用Java9编译和运行。请注意，与外部系统(connectors，文件系统，reporters)交互的某些组件可能无法工作，目前相应的项目可能不支持Java9。

发布说明

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

作者：Flink官方网站
翻译：伍翀（云邪）