Flink 1.11.0 发布，有哪些值得关注的新特性？

阿里妹导读：7 月 7 日，Flink 1.11.0 正式发布。历时近 4 个月，Flink 在生态、易用性、生产可用性、稳定性等方面都进行了增强和改善。Apache Flink PMC、阿里巴巴高级技术专家王治江，同时也是这个版本的 release manager 之一，将和大家一一分享，并深度剖析 Flink 1.11.0 带来了哪些让大家期待已久的特性，对一些有代表性的 feature 从不同维度解读。

首先在每个版本的规划初期，会从志愿者中选出 1-2 名作为 release manager。1.11.0 版本我作为中国这边的 release manager，同时还有一名来自 Ververica 的 Piotr Nowojski 作为德国方的 release manager，这在某种程度上也说明中国的开发者和贡献度在整个社区的占比很重要。

接下来会进行这个版本的 feature kickoff。在一些大的方向上，社区的规划周期可能比较久，会分阶段、分步骤跨越多个版本完成，确保质量。每个版本的侧重点也会有所不同，比如前两个版本侧重于批处理的加强，而这个版本更侧重于流处理易用性的提升。社区规划的 feature 列表会在邮件列表中发起讨论，以收集更多的用户/开发者意见和反馈。

一般的开发周期为 2-3 个月时间，提前会明确规划出大概的 feature freeze 时间，之后进行 release candidate 的发布和测试、以及 bug fix。一般经过几轮的迭代周期后会正式投票通过一个相对稳定的 candidate 版本，然后基于这个版本正式发布。

CREATE TABLE my_table (  ...) WITH (  'connector'='...', -- e.g. 'kafka'  'format'='debezium-json',  'debezium-json.schema-include'='true' -- default: false (Debezium can be configured to include or exclude the message schema)  'debezium-json.ignore-parse-errors'='true' -- default: false);

• Hive 写入：FLIP-115 完善扩展了 FileSystem connector 的基础能力和实现，Table/SQL 层的 sink 可以支持各种格式（CSV、Json、Avro、Parquet、ORC），而且支持 Hive table 的所有格式。

• Partition 支持：数据导入 Hive 引入 partition 提交机制来控制可见性，通过sink.partition-commit.trigger 控制 partition 提交的时机，通过 sink.partition-commit.policy.kind 选择提交策略，支持 SUCCESS 文件和 metastore 提交。

• Hive 读取：实时化的流式读取 Hive，通过监控 partition 生成增量读取新 partition，或者监控文件夹内新文件生成来增量读取新文件。

• FLIP-123 通过 Hive Dialect 为用户提供语法兼容，这样用户无需在 Flink 和 Hive 的 CLI 之间切换，可以直接迁移 Hive 脚本到 Flink 中执行。

• 提供 Hive 相关依赖的内置支持，避免用户自己下载所需的相关依赖。现在只需要单独下载一个包，配置 HADOOP_CLASSPATH 就可以运行。

• 对用户而言，在 Flink 中改造已有的 source 或者重新实现一个生产级的 source connector 不是一件容易的事情，具体体现在没有公共的代码可以复用，而且需要理解很多 Flink 内部细节以及实现具体的 event time 分配、watermark 产出、idleness 监测、线程模型等。

• 批和流的场景需要实现不同的 source。

• partitions/splits/shards 概念在接口中没有显式表达，比如 split 的发现逻辑和数据消费都耦合在 source function 的实现中，这样在实现 Kafka 或 Kinesis 类型的 source 时增加了复杂性。

• 在 runtime 执行层，checkpoint 锁被 source function 抢占会带来一系列问题，框架很难进行优化。

• 首先在 Job Manager 和 Task Manager 中分别引入两种不同的组件 Split Enumerator 和 Source reader，解耦 split 发现和对应的消费处理，同时方便随意组合不同的策略。比如现有的 Kafka connector 中有多种不同的 partition 发现策略和实现耦合在一起，在新的架构下，我们只需要实现一种 source reader，就可以适配多种 split enumerator 的实现来对应不同的 partition 发现策略。

• 在新架构下实现的 source connector 可以做到批流统一，唯一的小区别是对批场景的有限输入，split enumerator 会产出固定数量的 split 集合并且每个 split 都是有限数据集；对于流场景的无限输入，split enumerator 要么产出无限多的 split 或者 split 自身是无限数据集。

• 复杂的 timestamp assigner 以及 watermark generator 透明的内置在 source reader 模块内运行，对用户来说是无感知的。这样用户如果想实现新的 source connector，一般不再需要重复实现这部分功能。

• 每次调用可以处理 N 条数据。

• 数据格式基于 Apache Arrow，大大降低了 Java、Python 进程之间的序列化/反序列化开销。

• 方便 Python 用户基于 Numpy 和 Pandas 等数据分析领域常用的 Python 库，开发高性能的 Python UDF。

• PyFlink table 和 Pandas DataFrame 之间无缝切换（FLIP-120），增强 Pandas 生态的易用性和兼容性。

• Table & SQL 中可以定义和使用 Python UDTF（FLINK-14500），不再必需 Java/Scala UDTF。

• Cython 优化 Python UDF 的性能（FLIP-121），对比 1.10.0 可以提升 30 倍。

• Python UDF 中用户自定义 metric（FLIP-112），方便监控和调试 UDF 的执行。

• Session 模式：提前启动一个集群，所有作业都共享这个集群的资源运行。优势是避免每个作业单独启动集群带来的额外开销，缺点是隔离性稍差。如果一个作业把某个 Task Manager（TM）容器搞挂，会导致这个容器内的所有作业都跟着重启。虽然每个作业有自己独立的 Job Manager（JM）来管理，但是这些 JM 都运行在一个进程中，容易带来负载上的瓶颈。

• Per-job 模式：为了解决 session 模式隔离性差的问题，每个作业根据资源需求启动独立的集群，每个作业的 JM 也是运行在独立的进程中，负载相对小很多。

• 上游输出端控制单个 sub partition 堆积 buffer 的最大阈值（backlog），避免负载不均场景下单个链路上堆积大量 buffer。

• 在不影响网络吞吐性能的情况下合理修改上下游默认的 buffer 配置。

• 上下游数据传输的基础协议进行了调整，允许单个数据链路可以配置 0 个独占 buffer 而不死锁，这样总的 buffer 数量和作业并发规模解耦。根据实际需求在吞吐性能和 checkpoint 速度两者之间权衡，自定义 buffer 配比。

• Checkpoint barrier 跨数据 buffer 传输，不在输入输出队列排队等待处理，这样就和算子的计算能力解耦，barrier 在节点之间的传输只有网络延时，可以忽略不计。

• 每个算子多个输入链路之间不需要等待 barrier 对齐来执行 checkpoint，第一个到的 barrier 就可以提前触发 checkpoint，这样可以进一步提速 checkpoint，不会因为个别链路的延迟而影响整体。

• 为了和之前 aligned checkpoint 的语义保持一致，所有未被处理的输入输出数据 buffer 都将作为 channel state 在 checkpoint 执行时进行快照持久化，在 failover 时连同 operator state 一同进行恢复。换句话说，aligned 机制保证的是 barrier 前面所有数据必须被处理完，状态实时体现到 operator state 中；而 unaligned 机制把 barrier 前面的未处理数据所反映的 operator state 延后到 failover restart 时通过 channel state 回放进行体现，从状态恢复的角度来说最终都是一致的。注意这里虽然引入了额外的 in-flight buffer 的持久化，但是这个过程实际是在 checkpoint 的异步阶段完成的，同步阶段只是进行了轻量级的 buffer 引用，所以不会过多占用算子的计算时间而影响吞吐性能。