Apache Flink 1.16正式发布

沈雅

Apache Flink持续保持高速发展，是Apache最活跃的社区之一。Flink 1.16共有240多个Contributor热情参与，共完成了19个FLIP和1 100多个issue，给社区带来非常多振奋人心的功能。

Flink已经是流计算领域的领跑者，流批一体的概念逐渐得到大家的认可，并在越来越多的公司成功落地。之前的流批一体更强调统一的API和统一的计算框架。2022年，在此基础上，Flink推出了Streaming Warehouse，进一步升级了流批一体的概念：真正完成了流批一体的计算和流批一体的存储的融合，从而实现流批一体的实时化分析。

在1.16版本里，Flink社区对流、批都完成了众多改进：在批处理方面，完成了易用性、稳定性、性能提升全方位的改进，1.16是Fink批处理的里程碑式的版本，是走向成熟的重要一步。

易用性，引入SQL Gateway并完全兼容HiveServer2，用户可以非常方便的提交Flink SQL作业和Hive SQL作业，同时也很容易连接到原有的Hive生态。

功能：Flink SQL用户支持通过Join Hint指定Join策略，避免不合理的执行计划；Hive SQL的兼容性已经达到94 %，用户可以以极低的成本完成Hive到Flink的迁移。

稳定性，通过预测执行减少作业长尾以提高作业整体运行稳定性；支持自适应Hash Join，通过失败回滚机制避免作业失败。

性能提升，对多分区表进行动态分区裁剪以提高处理效率，TPC-DS在10 TB规模数据集下性能提升了30 %；支持混合Shuffle模式，提高资源使用率和处理性能。

在流处理方面，也完成了很多重大改进：

Changelog State Backend可以为用户提供秒级甚至毫秒级Checkpoint，从而大幅提升容错体验，同时为事务性Sink作业提供更小的端到端延迟体验。

维表关联在流处理中被广泛被使用，引入了通用的缓存机制加快维表查询速度，引入了可配置的异步模式提升维表查询吞吐，引入可重试查询机制解决维表延迟更新问题。这些功能都非常实用，解决了用户经常抱怨的痛点，支持了更丰富的场景。

从Flink SQL诞生第一天就存在一些非确定性操作，可能導致用户作业出现错误结果或作业运行异常，这给用户带来了极大的困扰。

随着流批一体的进一步完善和Flink Table Store的不断迭代，Flink社区正一步一步推动Streaming Warehouse从概念变为现实并走向成熟。

流式数仓

流式数仓（Streaming Warehouse）更准确地说，其实是make data warehouse streaming，就是让整个数仓所有分层的数据全部实时地流动起来，从而实现一个具备端到端实时性的纯流服务（Streaming Service），并且用一套统一API和计算框架来处理和分析所有流动中的数据。

批处理

得益于在流处理的长期投资，流处理已经成为流计算领域的领导者，在批处理上也投入了更多的精力，使其成为一个优秀的批处理引擎。流批处理统一的整体体验也将会更加顺畅。

SQL Gateway

从各个渠道反馈中了解到，SQL Gateway一直是用户非常期待的功能，尤其是对批用户。1.16里，该功能终于完成。SQL Gateway是对SQL Client的扩展和增强，支持多租户和插件式API协议（Endpoint），解决了SQL Client只能服务单用户并且不能对接外部服务或组件的问题。当前SQL Gateway已支持REST API和HiveServer2协议，用户可以通过cURL、Postman以及各种编程语言的HTTP客户端链接到SQL Gateway提交流作业、批作业，甚至OLAP作业。

Hive语法兼容

为了降低从Hive到Flink的迁移成本，这个版本里引入了HiveServer2协议并继续改进Hive语法的兼容性。

HiveServer2协议允许用户使用Hive JDBC/Beeline和SQL Gateway进行交互，Hive生态（DBeaver，Apache Superset，Apache DolphinScheduler，and Apache Zeppelin）也因此很容易迁移到Flink。当用户使用HiveServer2协议连接SQLGateway，SQL Gateway会自动注册Hive Catalog，自动切换到Hive方言，自动使用批处理模式提交作业，用户可以得到和直接使用HiveServer2一样的体验。

Hive语法已经是大数据处理的事实标准，Flink完善了对Hive语法的兼容，增加了对Hive若干生产中常用语法的支持。通过对Hive语法的兼容，可以帮助用户将已有的Hive SQL任务迁移到Flink，并且方便熟悉Hive语法的用户使用Hive语法编写SQL以查询注册进Flink中的表。到目前为止，基于Hive qtest测试集（包含12 000个SQL案例），Hive 2.3版本的查询兼容性已达到94.1 %，如果排除ACID的查询语句，则已达到97.3 %。

Join Hint

Hint一直是业界用来干预执行计划以改善优化器缺点的通用解决方案。Join作为批作业中最广泛使用的算子，Flink支持多种Join策略。统计信息缺失或优化器的代价模型不完善都会导致选出错误Join策略，从而导致作业运行慢甚至有运行失败的风险。用户通过指定Join Hint，让优化器尽可能选择用户指定的Join策略，从而避免优化器的各种不足，以确保批作业的生产可用性。

自适应Hash Join

对于批作业而言，数据倾斜是非常常见的，而此时使用Hash Join可能运行失败，这是非常糟糕的体验。为了解决该问题，引入了自适应的Hash Join：Join算子运行时一旦Hash Join运行失败，可以自动回退到Sort Merge Join，并且是Task粒度。通过该机制可确保Hash Join算子始终成功，从而提高了作业的稳定性。

批处理的预测执行

为了解决问题机器导致批作业处理慢的问题，Flink 1.16引入了预测执行。问题机器是指存在硬件问题、突发I/O繁忙或CPU负载高等问题的机器，这些问题可能会使得运行在该机器上的任务比其他机器上的任务要慢得多，从而影响批处理作业的整体执行时间。

当启用预测执行时，Flink将持续检测慢任务。一旦检测到慢任务，该任务所在的机器将被识别为问题机器，并通过黑名单机制被加黑。调度器将为慢任务创建新的执行实例并将它们部署到未被加黑的节点，同时现有执行实例也将继续运行。新的执行实例和老的执行实例将处理相同的输入数据并产出相同的结果数据。一旦任何执行实例率先完成，它将被视为该任务的唯一完成执行实例，并且该任务的其余执行实例都将被取消。

大多数现有Source都可以使用预测执行。只有当一个Source使用了SourceEvent时，它必须额外实现Supports Handle Execution Attempt Source Event接口以支持预测执行。目前Sink尚不支持预测执行，因此预测执行不会在Sink上发生。

Web UI和REST API也有了改进，以显示任务的多个执行实例和被加黑的TaskManager。

混合Shuffle模式

为批处理引入了一种新的混合Shuffle模式，它结合了Blocking Shuffle和Pipeline Shuffle（主要用于流式处理）的优点。与Blocking Shuffle一样，它不要求上下游任务同时运行，这允许使用很少的资源执行作业；与Pipeline Shuffle一样，它不要求上游任务完成后才执行下游任务，这在给定足够资源情况下减少了作业的整体执行时间。

用户可以选择不同的落盘策略，以满足减少数据落盘或是降低任务重启代價的不同需求。

注意：该功能为实验性的，并且默认关闭。

Blocking Shuffle进一步改进

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

动态分区裁剪

对于批作业，生产环境中分区表比非分区表使用更为广泛。当前Flink已经支持静态分区裁剪，即在优化阶段，优化器将Filter中的Partition相关的过滤条件下推到Source Connector中从而减少不必要的分区读取。星形模型是数据集市模式中最简单且使用最广泛的模式，很多用户的作业没法使用静态分区裁剪，因为分区裁剪信息在执行时才能确定，这就需要动态分区裁剪技术，即运行时根据其他相关表的数据确定分区裁剪信息，从而减少对分区表中无效分区的读取。通过TPC-DS 10 TB规模数据集的验证，该功能可提升30 %的性能。

流处理

在1.16中Checkpoint、SQL、Connector和其他领域都进行了改进，从而确保Flink在流计算领域继续领先。

Changelog State Backend旨在令Checkpoint的间隔更短、更加可预测。这个版本在自身易用性上和与其他State Backend兼容性上做了诸多改进，使其达到生产可用。

对于使用Flink构建的云服务应用来说，Rescaling是一种非常频繁的操作。这个版本使用了RocksDB的区间删除来优化增量RocksDB State Backend的Rescaling性能。区间删除被用来避免在Rescaling过程中大量的扫描和单点删除操作，对有大量的状态需要删除的扩并发来说，单个并发上的恢复速度可以提高2～10倍。

改善State Backend的监测体验和可用性

这个版本还改善了状态后台的监控体验和可用性。之前，RocksDB的日志位于它自己的DB目录中，这使得调试RocksDB没那么容易。这个版本让RocksDB的日志默认留在Flink的日志目录中，新增了RocksDB相关的统计指标，以帮助调试DB级别的性能，例如，在DB内的总块缓存命中/失败计数。

支持透支缓冲区

透支缓冲区（Overdraft Buffers）旨在缓解反压情况下Subtask被阻塞的概率，可以通过设置taskmanager.network. memory.max-overdraft-buffers-per-gate开启。

从1.16开始，一个Flink的Subtask可以申请5个（默认）额外的透支缓冲区。透支缓冲区会轻微地增加作业的内存使用量，但可以极大地减少Checkpoint的间隔，特别是在开启Unaligned Checkpoint情况下。只有当前Subtask被下游Subtasks反压且当前Subtask需要请求超过1个网络缓冲区（Network Buffer）才能完成当前的操作时，透支缓冲区才会被使用。

对齐Checkpoint超时

这个版本更新了从Aligned Checkpoint（AC）切换到Unaligned Checkpoint（UC）的时间点。在开启UC的情况下，如果配置了execution.checkpointing.aligned-checkpointtimeout，在启动时每个Checkpoint仍然是AC，但当全局Checkpoint持续时间超过aligned-checkpoint-timeout时，如果AC还没完成，那么Checkpoint将会转换为UC。

以前，对一个Substask来说，AC到UC的切换需要等所有上游的Barriers到达后才能开始，在反压严重的情况下，在checkpointing-timeout过期之前，下游的Substask可能无法完全地收到所有Barriers，从而导致Checkpoint失败。

在这个版本中，如果上游Subtask中的Barrier无法在execution.checkpointing.aligned-checkpoint-timeout内发送到下游，Flink会让上游的Subtask先切换成UC，以把Barrier发送到下游，从而减少反压情况下Checkpoint超时的概率。

流计算的非确定性

Flink SQL用户经常抱怨理解流处理的成本太高，其中一个痛点是流处理中的非确定性（而且通常不直观），它可能会导致错误的结果或异常，而这些痛点在Flink SQL的早期就已经存在了。

对于复杂的流作业，现在可以在运行前检测并解决潜在的正确性问题。如果问题不能完全解决，一个详细的消息可以提示用户如何调整SQL，以避免引入非确定性问题。

维表增强

维表关联在流处理中被广泛使用，在1.16中为此加入了多项优化和增强：

支持通用的缓存机制和相关指标，可以加速维表查询；

通过作业配置或查询提示支持可配置的异步模式（ALLOW_UNORDERED），在不影响正确性的前提下大大提升查询吞吐；

可重试的查询机制让用户解决维表数据更新延迟问题有了更多的手段。

异步I/O支持重试

为异步I/O引入了内置的重试机制，它对用户现有代码是透明的，可以灵活地满足用户的重试和异常处理需求。

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提供了更全面的支持，支持了旁路输出、Broadcast State等功能，并完善了对于窗口功能的支持。在Python DataStream API中，添加了对于更多的Connector以及Format的支持，例如添加了对于Elasticsearch，Kinesis，Pulsar，Hybrid Source等Connector的支持以及对于Orc，Parquet等Format的支持。有了这些功能之后，Python API已经基本对齐了Java和Scala API中绝大部分的重要功能，用户已经可以使用Python语言完成大多数类型Flink作业的开发。

DataStream中的缓存

支持通过DataStream#cache缓存Transformation的执行结果。缓存的中间结果在首次计算中间结果时才生成，以便以后的作业可以重用该结果。如果缓存丢失，原始的Transformation将会被重新計算以得到结果。目前该功能只在批处理模式下支持。这个功能对于Python中的ML和交互式编程非常有用。

History Server及已完成作业的信息增强

在这个版本中加强了查看已完成作业的信息的体验。

JobManager / HistoryServer WebUI提供了详细的执行时间指标，包括任务在每个执行状态下的耗时，以及在运行过程中繁忙/空闲/反压总时间。

Protobuf格式

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

为异步Sink引入可配置的RateLimitingStrategy

1.15中实现了异步Sink，允许用户轻松实现自定义异步 Sink，新版本里此进行扩展以支持可配置的RateLimiting Strategy。这意味着Sink的实现者现在可以自定义其异步Sink在请求失败时的行为方式，具体行为取决于特定的Sink。如果没有指定RateLimitingStrategy，它将默认使用AIMDScalingStrategy。