实时分析的变革力量：下一代流存储准备好了吗？

1. Kafka 在实时分析场景遇到的问题；

5. Fluss 开源。

在技术方面，大数据计算架构经历了显著的演变。从最初的 Hive 传统数据仓库，到引入 Lakehouse 湖仓架构，再到目前国内流行的 Paimon 流式湖仓架构，这些演进的核心驱动力在于提升业务的时效性。从传统的 T+1 天模式，逐步缩短到 T+1 小时，再到 T+1 分钟。然而，由于湖存储架构是基于文件系统的，其分钟级延迟几乎是极限。但是许多业务场景，如搜索推荐、广告归因和异常检测，都要求秒级的实时响应。因此，业界亟需能够支持秒级存储的解决方案。尽管大数据技术已经取得了长足的发展，但在大数据分析场景中，仍然缺乏一款能够有效支持秒级存储的解决方案。

那么在大数据里面最常用的秒级存储是什么呢？当然是 Apache Kafka。Flink 与 Kafka 的组合也已经成为业界构建实时数仓的典型架构。然而，这个组合在实际应用中并不总是那么理想，原因在于当我们将 Kafka 应用于大数据分析时，会遇到一系列挑战和问题。

一个主要的问题是，Kafka 不支持数据更新功能。在数据仓库中，“更新”是一个非常重要的功能，对于一个数仓来说，经常需要“更新”的能力去修正一些数据。由于 Kafka 不支持更新，所以它只能将主键上重复的数据都存储下来。当计算引擎消费这些数据时，就会接收到重复的数据。

为了确保计算结果的准确性，计算引擎必须执行去重操作。然而，这个去重过程本身是非常耗费资源的。在 Flink 中，这需要使用 State 来物化上游的全部数据，并且每次消费 Kafka 数据时，都必须承担去重的成本，这个成本是相当高的。这种高成本的去重要求限制了 Kafka 数据的业务复用能力。例如，在淘天集团构建实时数据中间层的过程中，由于 Kafka 的这些限制，他们选择不构建 DWS 层。

第二个主要问题是，Kafka 不支持数据探查功能。在数据仓库建设中，数据探查是一个基本能力。无论是排查问题还是进行数据探索，都需要进行数据查询。然而，Kafka 本质上是一个黑盒，不支持直接查询。为了解决这个问题，业界通常采用两种方案：

1. 同步到 OLAP 系统：将 Kafka 数据同步到 OLAP 系统中进行查询。不过，这种方法会引入额外的系统组件，增加复杂性和成本。此外，数据在不同系统间的同步也可能导致不一致性。

2. 使用 Trino 等查询引擎直接查询 Kafka：这种方法避免了数据同步问题，但由于 Kafka 仅支持 Full Scan，无法实现 Data Skipping，因此在处理大规模数据时效率较低。例如，在 1GB 数据上进行简单查询都可能需要一分钟，这使得这种方法在大规模应用中基本上不可行。

第三个问题是数据回溯的困难。在数据仓库中，数据回溯是常见需求，例如在物流行业中，可能需要回溯几个月的数据进行分析。然而，在 Kafka 中，长时间存储大量数据会导致成本过高，因此通常只能存储几天的数据。此外，当进行大规模数据回溯时，所有数据流量都必须经过 Kafka Broker，这会导致回溯操作的性能非常慢。同时，这种操作还会消耗 Broker 的 CPU 资源，污染其页面缓存（page cache），从而对其他在线业务产生负面影响。

最后一个问题是网络成本。根据多项数据资料显示，网络成本占据了 Kafka 成本的 88%。在数据仓库中，一写多读是非常常见的操作模式，并且每个消费者通常只消费数据的一部分。例如，在阿里巴巴内部的数万条 Flink SQL 作业中，平均每个作业仅使用了上游数据的 49% 的列。然而，当用户需要消费这 49% 的列时，仍然需要读取所有列的数据，这意味着需要承担 100% 的网络带宽成本。这种情况导致了网络资源的极大浪费。

总结来说，将 Kafka 用于实时分析场景时，会面临以下核心问题：不支持更新、无法探查、数据回溯难、网络成本高。这些问题导致 Flink + Kafka 的组合在某些实时分析应用场景中并不是最理想的选择。

那么其本质的原因是什么？

这是因为Kafka 是为流消息设计的，并不是为流分析设计的。每个系统都有其特定的定位和优势，Kafka 在消息队列场景中非常高效，因为它通常以行存格式（如 CSV、JSON、AVRO）存储数据。然而，对于需要处理大规模数据和复杂查询的分析场景来说，行存格式的效率则显得不足。需要底层存储具备强大的Data Skipping 能力，以及支持列裁剪和条件下推等特性。在这种情况下，列存格式显然更为适合。

在构建这样的四象限矩阵时，我们可以观察到一个有趣的现象：象限左边是业务型系统，右边是分析型系统，上面是流存储，下面是表存储。可以看到，业务型系统里面不管是数据库，还是流存储，都采用的是行存，因为行存在这个场景更为高效。相反，像 Iceberg, Snowflake 这些分析型系统都采用的列存，因为列存在分析场景更高效。在这个矩阵中，右上角是一个空白区域，代表这个市场里空缺了一个存储，即面向分析场景的流存储，不出意料的话，这个流存储采用的会是列存格式。

"FLink Unified Streaming Storage"，取了项目名的首字母缩写，拼成了 Fluss 这个单词。值得一提的是，Flink 这个名字源自德语，意为“敏捷迅速”，而 Fluss 恰巧也是个德语单词，意为“河流。这种命名不仅向 Flink 项目的起源致敬，也象征着流数据如同河流般源源不断地流动、分发，并最终汇聚到数据湖中。

接下来介绍一下 Fluss 的一些核心特性：

首先，不出所料，Fluss 采用列式的流存储。在底层文件存储中采用了 IPC Streaming Format 协议，而 Arrow 是一种非常优秀的流式列存储格式。基于 Arrow，我们实现了非常高效的列裁剪功能。右侧展示了对 Fluss 和 Kafka 的基准测试结果。横轴表示读取列的数量，纵轴表示读取吞吐量。可以看到，随着裁剪的列数增加，Fluss 的读取性能成比例上升。当裁剪到 90% 的列时，Fluss 的读取吞吐量已经提高了 10 倍。此外，Fluss 的列裁剪是在服务端进行的，这意味着发送给客户端的数据已经是裁剪过的，从而节省了大量的网络成本。

此外，KV 生成的 Changelog 可以直接被 Flink 流读取，无需额外的去重操作，节省了大量计算资源，实现了数据的业务复用。由于我们构建了 KV 索引，因此可以支持高性能的主键点查，并可作为实时处理链路中的维表关联。用户还可以通过点查的 query 语句直接探查 Fluss 数据，我们还支持 LIMIT、COUNT 等查询功能，以满足用户的数据探查需求。

在底层，Fluss 维护了一个 Compaction Service，该服务会自动地将 Fluss 数据转换为湖存储的格式，并确保两边元数据的一致性。此外，它还保证两边的数据分布也是一致的，即分区和分区一一对齐，Bucket 和 Bucket 也一一对齐。这使得在流转湖的过程中，无需引入网络 Shuffle，只需将 Arrow 文件直接转换为 Parquet 文件即可。这种转换在业界已有非常成熟且高效的实现。

在拥有湖和流两层数据后，Fluss 的一个关键特性是共享数据。具体来说，湖存储作为流存储的历史数据层，负责存储长周期、分钟级延迟的数据；而流存储作为湖存储的实时数据层，负责存储短周期、毫秒级延迟的数据，这两者的数据可以互相共享。当进行流读取时，湖存储可以作为历史数据提供高效的回溯性能。在回溯到当前位点后，系统会自动切换到流存储继续读取，并确保不会读取重复数据。在批查询分析中，流存储可以为 Lakehouse 提供实时数据的补充，从而实现 Lakehouse 秒级新鲜度的分析。我们将这种功能称为 Union Read。

除此之外，我们同步到湖存储的格式完全遵循现有湖存储的开放协议，因此现有的一些查询引擎（如 Spark、StarRocks、Trino）可以直接查询湖存储中的数据，无缝融入用户已有的 Lakehouse 架构中。目前，Fluss 已经完成了对 Paimon 的完全集成，对 Iceberg 的集成也在计划中。

这就是我们Fluss整体的架构图，Fluss是一个面向实时分析的流存储。Fluss 需要维护一个 Server 集群，提供实时读写的能力，同时使用 Remote Storage 来做数据的分层，降低数据存储成本。并且跟Lakchouse 做了一个非常无缝的集成来支持丰富的查询能力。Fluss 的核心特性包括实时的流读流写、列式裁剪、流式的更新、CDC订阅、实时点查、还有湖流一体。

Fluss 的核心特性结合，实现了一个非常理想的应用场景 Delta Join。在 Flink 中，双流 Join 是一个非常基础的功能，常用于构建宽表。然而，这也是一个常常让开发人员感到头疼的功能。因为双流 Join 需要在 State 中维护上游全量的数据，这导致其状态通常非常庞大。例如，淘宝最大的 Flink 作业之一是成交引导的双流 Join（曝光关联订单），需要消耗 50TB 的状态。但这带来了很多问题，包括成本高、作业不稳定、Checkpoint超时、重启恢复慢等等。

因此我们充分利用 Fluss 的 CDC 流读+索引点查的能力研发了一套新的 Flink 的 Join 算子实现，叫 Delta Join。Delta Join 可以简单理解成“双边驱动的维表Join”，就是左边来了数据，就根据Join Key去点查右表；右边来了数据，就根据 Join Key 去点查左表。全程就像维表Join一样不需要state，但是实现了双流Join一样的语义，即任何一边有数据更新，都会触发对关联结果的更新。

在测试中，我们使用了淘宝最大的双流 Join 作业进行性能评估。在从双流 Join 迁移到 Delta Join 后，成功减免了 50TB 的大状态，使得作业运行更加稳定，Checkpoint 也不再超时。在双十一的数据压测回追中，我们发现，在保证相同吞吐量的情况下，Flink 的资源消耗能够降低10倍，从2300 CU 减少到200 CU。此外，在回追过程中，我们还可以利用湖流一体归档的 Paimon 表加上 Flink Batch Join 进行数据回追，将回追1天数据的时间从4小时缩短到0.5小时。使用批作业进行数据回追，展示了流批一体的一个非常有前景的应用场景。

除了资源的减少和性能的提升，对于用户最大的收益其实是灵活性的提升。以前的 State 是 Flink 内置的黑盒，用户看不见摸不着，一修改作业就要重跑 State，耗时耗力。在使用 Delta Join 后，相当于状态与作业进行了解耦，修改作业不需要重跑 State，所以回追很高效。并且数据都在 Fluss 里面，变得可查可分析，提升了业务灵活性和开发效率。目前，我们已经在 Flink 社区提交了 Delta Join 的 FLIP-486 提案，对于这个提案感兴趣的朋友可以关注一下。

关于 Fluss 的未来规划，最重要的有三件事，这三件事分别对应了 Fluss 与三个开源软件之间的关系：

1. Kafka 协议兼容：这是为了帮助已有的流数据更好地迁移到 Fluss 上。

2. 与 Flink 的深度协同优化：这一规划包括通过存储+优化器+执行引擎的协同优化，以解决之前存在的一些难点和痛点。Delta Join 就是一个很好的例子，通过这种深度协同，Fluss 可以与 Flink 紧密结合，提升整体的流处理性能和稳定性。

3. 为 Paimon 提供实时数据层：通过打造湖流一体架构，Fluss 希望与 Paimon 结合，提供一个实时与离线一体化的存储解决方案。

Fluss 目前已经在 GitHub 上以 Apache 2.0 协议正式开源。

项目地址为：https://github.com/alibaba/fluss，欢迎大家关注和 Star。并且我们计划于 2025 年将其捐赠到 Apache 软件基金会。在此，我们诚挚地邀请各位加入 Fluss 开源社区，共同促进这一新兴项目的成长与发展。欢迎大家加入 Fluss 社区钉钉群109135004351，欢迎大家一起探索，参与开发和贡献，并携手构建下一代的流存储技术！