流批一体技术简介_flink 流批一体

摘要：本文由阿里云 Flink 团队苏轩楠老师撰写，旨在向 Flink 用户整体介绍 Flink 流批一体的技术和挑战。内容主要分为以下三个部分：

1. 流批一体技术简介

2. 面临的挑战

3. 总结

Flink 的流批一体概念相信大家并不陌生，尽管流批一体被广泛讨论，但在很多使用者心中，Flink 更多的是作为流计算引擎的事实标准，常有朋友提到 Flink，也是更熟悉其在流计算场景的使用。今天，本文旨在全面向读者介绍 Flink 的流批一体技术及其所面临的技术挑战。在未来的分享中，我们将进一步探讨流批一体在不同场景下的应用，以及通过公开渠道收集到的一些企业使用 Flink 流批一体功能的落地情况。

一、流批一体技术简介

在这一章节我们会简要地介绍一下流批一体相关的技术，让大家了解流批一体的整体架构以及架构中各个组件的技术选择。

1. 流批一体的架构演进

首先我们来看看，流批一体的架构是怎样演进而来的。

（1）Lambda 架构

Lambda 架构通常是在原有的离线计算中发展而来的。在离线计算的链路（批处理层）上，为了满足业务数据实时性的要求，会在离线链路的基础上再增加一条实时计算链路（速度处理层）。最后，在对外提供服务的时候，会合并批处理层和速度层的视图（服务层）。虽然 Lambda 架构可以在不改变原有离线计算架构的基础上，同时享受到离线和实时计算带来的好处，但是它在使用的过程中存在着以下的问题：

• 维护离线和实时两套系统的运维成本成本高
• 需要为离线和实时开发两套代码，学习成本和开发成本高
• 离线和实时的计算引擎不同，数据一致性难以保证
• 使用存储格式不同，数据管理更加复杂

图片来源：Questioning the Lambda Architecture [1]

（2）Kappa 架构

Kappa 架构相当于在 Lambda 架构上去掉了批处理层（Batch Layer），只留下单独的流处理层（Speed Layer）。通过消息队列的数据保留功能，来实现上游重放（回溯）能力。Kappa 架构解决了上面提到的 Lambda 架构所面临的问题，但是在这个架构下，就无法享受到离线计算带来的好处了。比如通过流计算来回溯的性能会比使用批计算的性能要差。而且对于没有实时需求的作业，使用 Kappa 架构的流计算会造成不必要的资源浪费。

图片来源：Questioning the Lambda Architecture [1]

（3）流批一体架构

通过使用流批一体的计算引擎和流批一体的存储格式，我们可以很好地解决 Lambda 和 Kappa 架构中存在的问题。

• 在流批一体的架构中，我们使用流批一体的计算引擎可以避免维护两套系统的运维成本。
• 使用相同的流批一体的存储格式，可以避免分别为离线链路和实时链路使用两套不同的存储，减少了存储链路的冗余和成本。
• 用户只需要写一套代码就能同时用于实时计算和离线计算，大大降低了用户的学习成本和开发成本。同时，使用统一的计算引擎，统一的代码可以更好地保证数据的一致性。

图片来源：Flink Forward Asia 2020 - 流批一体技术在天猫双11的应用 [2]

介绍完流批一体的架构之后，让我们来看一下流批一体架构中最重要的两个组件：计算引擎和存储。

2. 流批一体的计算引擎

Apache Flink 从设计之初就提出了“批处理是流处理的特殊情况”。用户可以使用 DataStream API 和 Flink 的 SQL API 来同时定义流作业和批作业。Flink 在流批一体这个方向上已经做了非常多的工作来提高用户的体验，作业的稳定性和性能，目前 Flink Batch 已经在很多公司的生产环境上落地。Flink 社区未来也会持续投入 Flink 流批一体的发展。

Apache Spark 也是最早提出流批一体理念的计算引擎之一，可以用作流批一体计算引擎。与 Flink 不同的是，它的流计算是基于微批（mini-batch）来实现的，在流计算语义的支持和端到端延迟上会差一些，面对复杂、大规模实时计算场景的极致需求可能会力不从心。Apache Spark 虽然也在探索使用 Continuous Processing 来支持流计算，降低延迟，但目前还是属于实验阶段 [3]，并且经过笔者的观察 Continuous Processing 的投入一直不大，2021 年之后就几乎停滞了。

3. 流批一体的湖表格式

在流批一体这个大场景下，计算引擎只是其中的一环，流批一体的存储格式更是不可或缺的一部分。Flink 在流批一体的存储格式上做了许多探索，对接了多个不同的存储格式，目前在开源社区主流的支持 Flink 流批一体的存储格式有下面这些：

Apache Paimon 是流批一体的湖存储格式。可以使用 Flink CDC 来一键入湖到 Paimon 中，也可以通过 Flink SQL 或 Spark SQL 来批写、流写到 Paimon 当中。Paimon 也可以被 Flink 或 Spark 流读，这也是它作为流式数据湖的特有能力之一。它有着强大的流读流写支持，给流式湖存储带来仅 1-5 分钟的延迟 [4]。

Apache Hudi 原生支持多引擎，因此既可以对批流进行读写消费，也可以使用Presto进行交互式分析 [5]。Flink 接入之后，把 Hudi 的时延可以达到十分钟级[4]。

Apache Iceberg 早在 2020 年，阿里云就试图把 Flink 融入 Iceberg 中，在 Iceberg 中做了很多 Flink 的集成。在把 Flink 融入 Iceberg 后，Iceberg 就有了 Flink 流读流写的力。目前 Flink 写入 Iceberg，并不能太实时，因此更推荐在 1 小时左右的更新 SLA 保障[4]。

4. 流批一体的数仓

除了流批一体的湖表格式，还有流批一体的数仓也可以作为流批一体的存储，例如开源的 Starrocks, Clickhouse, Apache Druid 等。还有类似阿里云的商业化产品 Hologres 也可以作为流批一体的数仓。但是数仓的存储成本比湖存储更高，我们又看到一些的做法是把数据通过湖表格式写入到数据湖中，然后通过数仓来分析数据湖中的数据做 OLAP 分析，例如 Startrocks + Paimon[6], Hologres + Paimon [7] 等湖仓一体方案。

二、面临的挑战

在大家使用 Flink Batch 流批一体实践的过程中，难免会遇到各种各样的问题和挑战。Flink 社区也在积极地解决大家在使用过程中遇到的问题，对 Flink 的批作业能力进行打磨，使Flink 的流批引擎的能力逐步地完善。下面介绍了近些年来，大家在 Flink 流批一体实践中遇到的挑战，以及社区的解决方案。

1. 流批 Shuffle 差异

Flink 流作业的 shuffle 与 批作业的 shuffle 通常是不一样的，在流作业的情况下使用的是 Pipeline Shuffle，Pipeline Shuffle 的数据是不用落盘的，但是这要求作业启动的时候所有的算子都要启动起来，这与常见的 Batch 作业调度需求不匹配。因此，在批模式下，通常都是使用 Blocking Shuffle，这样上游 task 会把 Shuffle 数据写到离线文件中，等下游 task 启动以后，再来消费 Shuffle 的数据。

Flink 内部默认的实现是使用 Internal Shuffle， 是把上游计算节点数据写到 TaskManager 本地盘，下游节点连接到上游 TaskManager 上读取 Shuffle 文件。这会导致 TaskManager 计算工作完成以后，不能立刻退出，要等下游消费完 Shuffle 文件后才能释放掉。这样不仅造成了资源浪费，而且容错代价大。

因此，Flink 社区在已开始设计 Shuffle Service 的时候就把他作为一个 pluggable [8]，以便于用户能够方便的拓展来实现 Remote Shuffle Service。Remote Shuffle Service 通过单独的集群提供数据的 Shuffle 服务，可以避免 TaskManager Shuffle 的资源利用率低和容错开销大的问题。目前 Apache Celeborn[9] 支持作为 Flink 的 remote shuffle.

图片来源：Flink Forward Asia 2022 - Flink Shuffle 3.0 Vision, Roadmap & Progress [10]

同时社区也在推动 Shuffle 3.0 中提出了 Hybrid Shuffle，Hybrid Shuffle 将流式 Pipeline Shuffle 跟批式 Blocking Shuffle 的特点结合在一起，让用户在写数据时，既可以写入内存通过内存直接进行消费，也可以在内存中存放不下这么多数据、下游消费不够及时的时候，将数据写入到磁盘当中进行后期消费。通过自适应切换，在上游产出数据的过程中和完成后，下游可以随时消费，从而彻底消除资源碎片的情况 [10]。

2. Batch 性能

Flink batch 作业的性能也是用户在使用 Flink batch 中最关心的问题之一，Flink batch 在作业性能优化上面做了非常多的改进，使得 Flink batch 在 TPC-DS benchmark 上的表现在每个版本都有很大的提升。例如使用 Operator Fusion Codegen [11] 来优化 SQL planner 生成的代码，通过 adaptive local hash aggregate[12] 来动态决定是否使用 local aggregation，通过 runtime filter 和 dynamic data prune 来优化数据处理的效率，实现 Adaptive Execution Plan（AQE） 做了自动并发推断，动态负载均衡等[13]。

图片来源：Flink Forward Asia 2023 - 深入解析 Flink Batch 新进展 [13]

3. 慢节点问题

在一个分布式系统里，因为个别的机器故障、资源紧张或者是网络问题，可能导致单个并发的性能下降，这些慢的节点可能成为整个作业的瓶颈。和传统 MapReduce、Spark 的思路类似，Flink 1.17 版本引入了推测执行来解决慢节点的问题 [14]。当检测到长尾任务后，在非热的机器上部署长尾任务的镜像实例。哪个先执行完就用哪个结果，并把其他的镜像任务取消掉。

4. 并发配置易用性

在 Flink Batch 作业中，为作业节点设置适当的并行性并不是一件容易的事情。在批作业中，并行度设置得太小可能导致太长的执行时间和 Failover 发生时大量的回退。相反，并行度设置得过大，则可能导致资源浪费和任务部署和网络 Shuffle 更多的成本开销。为了解决这个问题，Flink 在 1.15 中引入了 Adaptive Batch Scheduler [15]，这能让 Flink 根据消费的数据量大小来自动决定作业节点的并发度，免去了用户需要手动调整作业并发度的烦恼。

5. Hive SQL 兼容性

由于 Flink SQL 使用的是标准的 ANSI SQL，并且 Hive SQL 与 ANSI SQL 语法差异较多。不少用户在迁移 Hive SQL 到 Flink SQL 上的时候会遇到不少的阻碍。虽然 Flink SQL 本身提供了 Hive Dialect [15]，但是在 Flink 1.15 版本，距离完全兼容 Hive SQL 仍然有不小的差距。比如，快手在选定了一批准备迁移的作业后，通过解析验证，就发现了诸多不支持的语法。在快手给出 input 后，社区第一优先级做出了支持。如上图所示，我们列出了比较重要且很常用的一些语法，比如 CTAS、ADD JAR、USING JAR、宏命令、Transform 等。

Flink 社区在 1.16 版本做了大量工作, 包括 CTAS [16]、ADD JAR、USING JAR [17] 等等，来补全 Hive 语法。经过 qtest 测试，整体兼容度能达到 95%，基本能保证用户现有的 Query 都能迁到 Flink 上来。

三、总结

流批一体是 Flink 非常重要的一个发展方向之一，众多用户都在使用的过程中给予了 Flink 社区宝贵的反馈。随着更多的开发者加入了开源社区的开发工作，这让 Flink Batch 的能力不断得到打磨和发展。在社区成员的努力下，很多用户已经可以十分顺利地把 Flink 流批一体在他们的生产环境中落地。我们会在后期分享中，给大家介绍一些流批一体的主要落地场景，以及我们在公开渠道收集到的各个公司使用流批一体的落地情况。

虽然，Flink 流批一体已经达到生产可用的状态，但是社区也看到仍然有不少需要继续投入的地方，例如继续完善 DataStream API batch 的能力，使其能够完全与 DataSet API 能力对齐；更加深入地与 Apache Celeborn 结合，结合 Flink 特点实现动态切换 Shuffle 的机制，多级存储引入内存、支持 Flink Hybird Shuffle 等；更加深入地与 Apache Paimon 对接，整合流批一体引擎和存储的能力，让用户能够更加简单地使用 Apache Flink + Apache Paimon 搭建流批一体的数据湖仓。

[1] https://www.oreilly.com/radar/questioning-the-lambda-architecture/

[2] https://www.bilibili.com/video/BV1164y1o7yc/

[3] https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#continuous-processing

[4] https://flink-learning.org.cn/article/detail/d17cc1d2a06946b40c51d4301df6e540

[5] https://mp.weixin.qq.com/s?__biz=MzU0OTE4MzYzMw==&mid=2247560963&idx=2&sn=0993e780bf873dcec5a7eb6242ef4323

[6] https://flink-learning.org.cn/article/detail/02a574303b7e65fd53e13a82b40a8d8f

[7] https://flink-learning.org.cn/article/detail/84f501725034542a7f41e0670645c714

[8] https://cwiki.apache.org/confluence/display/FLINK/FLIP-31%3A+Pluggable+Shuffle+Service

[9] https://celeborn.apache.org/

[10] https://flink-learning.org.cn/article/detail/f6449048654123b163e29917e8ad5a79

[11] https://cwiki.apache.org/confluence/display/FLINK/FLIP-315+Support+Operator+Fusion+Codegen+for+Flink+SQL

[12] https://issues.apache.org/jira/browse/FLINK-30542

[13] https://developer.aliyun.com/ebook/8229/115382

[14] https://cwiki.apache.org/confluence/display/FLINK/FLIP-168%3A+Speculative+Execution+for+Batch+Job

[15] https://cwiki.apache.org/confluence/display/FLINK/FLIP-152

[16] https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=199541185

[17] https://cwiki.apache.org/confluence/display/FLINK/FLIP-214

欢迎大家加入 Flink Batch 交流钉钉群。本群旨在为 Flink Batch 爱好者提供一个交流技术和传递资讯的平台，在这里：

• 你可以掌握Flink Batch前沿的资讯，可以与 Flink 开发者及 Committer 面对面交流
• Flink Batch 的问题集中解决，各位开发者及 Committer 及时解决你的 Blocker

“Flink Batch 交流群”群的钉钉群号： 34817520