我们为什么从Kafka迁移至Pulsar？_大数据_StreamSQL团队

阿里、蚂蚁、晟腾、中科加禾精彩分享 AI 基础设施洞见，现购票可享受 9 折优惠｜AICon 了解详情 



 写点什么

最近，StreamSQL 在 Kafka 上构建了平台，但几经实践，研发团队认为 Kafka 并非最合适的工具。本文将阐述该团队用 Kafka 构建的解决方案，以及决定迁移到 Pulsar 的原因。

在今天，Apache Kafka 几乎是事件流的同义词。事件流是我们平台的核心部分，最近，我们将 Kafka 替换成了 Pulsar。我们已经与客户当面谈过这个问题，也在会议上讨论过。前不久，Apache Pulsar 社区的一个朋友建议我写一篇文章，分享我们的经验和转换的原因。

我们在 Kafka 上构建了平台，并且发现自己已经写了大量的代码来使系统按照我们的意愿来运行。我们认为，Kafka 并非适合这项工作的工具。这一结论显然对于很多用例来说，并不都正确，但即使这样，使用 Kafka 而不是 Pulsar 仍然是有意义的。我将在本文阐述我们用 Kafka 构建的解决方案，以及讲述决定迁移到 Pulsar 的原因。

问题陈述

StreamSQL 是什么？

StreamSQL 是一个围绕事件源构建的数据存储系统。StreamSQL 由三个组件组成：事件存储、转换和物化状态。事件存储是发送到我们系统的每个域事件不可篡改的账本。我们使用类似于 Cassandra、Redis 和 CockroachDB 的 API 为物化状态提供服务。Transformation （转换）是将事件映射到状态的纯函数。根据 Transformation，我们接收到的每个事件都被处理并应用到物化状态。

StreamSQL 在所有数据中追溯性地运行新的 Transformation。最终状态是整个事件流的真实物化。此外，你还可以通过回滚和回放事件来生成一个“虚拟”状态。虚拟状态可用于训练和验证机器学习模型，也可以用于调试目的（如用于前端开发的 Redux）。

要求

系统需要能够执行以下操作：

将每个域事件永久存储在系统中；
通过保证对每个传入的事件只处理一次，以保持物化状态的一致性；
能够按照我们接收到的相同顺序对所有历史事件进行 Transformation；
回滚并回放事件账本，并在该点物化视图。

最初的基于 Kafka 解决方案

最初的基于 Kafka 解决方案，由一组拼接在一起的大数据工具组成。系统将过去的事件存储在 S3 中，并用 Spark 对它们进行处理。对于流数据，它使用 Kafka 和 Flink。要保持事件和物化视图的一致性，需要在每个系统之间进行复杂的协调。

无限存储每个域事件

每个域事件都将通过 Kafka 进入系统，然后 Kafka 会将其保存到 S3 中。这就使得我们能够存储大量很少使用的数据，并且具有高持久性和低成本。

我们曾尝试在流上使用 Kafka 的无限保留，但是发现它不仅成本昂贵，而且难以维护。在更大的主题（Topic）上，我们开始看到性能下降和不稳定的延迟的现象。由于此时我们已经几乎完全迁移到 Pulsar 上，因此没有进一步研究原因。

从批数据引导物化视图

我们通过按顺序处理每个事件来物化视图。我们使用 Spark 来处理存储在 S3 的大部分历史数据。如果我们可以在这种情况发生时将事件暂停，事情就会变得简单了。在这种情况下，我们可以读取所有 S3 数据，然后切换到处理主题开头的 Kafka。实际上，从 Kafka 持久化到 S3 的事件之间有一个延迟，在将大型批集群交换为较小的流处理集群之间还存在另一个延迟。由于我们不能错过任何事件的处理，因此，我们使用 Spark 在 S3 中处理尽可能多的事件，然后让它返回最后一个事件的 ID。由于我们已经将 Kafka 配置为保留最近几周的数据，所以我们可以将 Kafka 的其余事件回填。

从 Kafka 回填

Spark 能够处理过去的大多数事件，但它并不能让我们了解最新的状态。为了处理最后一组过去的事件，我们已经配置了 Kafka 集群，以保留最后两周确认的事件。我们运行一个 Flink 作业来继续 Spark 启动的 SQL Transformation。我们将 Flink 指向 Kafka 中的第一个事件，并让它通读一遍，什么也不做，直到它到达 Spark 停止的 messageID 为止。从那时起，它将继续更新物化视图，直到它到达流的头部。最后，它通知 Transformermation API，物化视图是最新的，可以使用。

更新传入事件

一旦启动物化视图，StreamSQL 就必须保持物化视图是最新的。在这一点上，这个问题是微不足道的。Kafka 将每个传入事件直接传递给 Flink，然后 Flink 执行必要的更新。此时，Transformermation API 和 Spark 处于空闲状态。但是，我们仍然将每个传入的事件保存在 S3 中，以防用户更新或创建 Transformation。

多租户、回滚和回放、错误处理等

我们协调 Flink 和 Kafka 一起工作，保存物化视图的快照。通过适当的协调，我们可以实现无缝的回滚和回放功能。要对这一过程进行阐述，需要专门写一篇博文（我们希望在不久的将来会撰写）。

在本文中，我们也不会讨论如何扩展 Flink 和 Kafka 集群、如何处理服务故障，或者如何在所有这些不同的服务之间实现安全的多租户（提示：每个解决方案都有不同的答案）。

为什么是 Pulsar？

Pulsar 的构建是为了永久存储事件，而不是在系统之间传输事件。此外，Pulsar 是在 Yahoo! 为在全球范围内开发各种产品的团队服务之上构建的。它本身就支持地理分布和多租户。Pulsar 执行复杂的部署变得很容易，如为某些租户保留专用服务器。我们尽可能利用这些特性。这使得我们可以将大部分的自定义逻辑交给 Pulsar 去处理。

S3 中的分层存储

StreamSQL 用户可以随时创建新的物化视图。这些视图必须是所有事件的投影。因此每个 Transformation 都按顺序处理每个历史事件。在基于 Kafka 的解决方案中，我们将所有已确认的事件流传输到 S3 或 GCS。然后，Spark 中的批处理管道处理这些事件。整个系统需要我们协调事件流、批处理存储、批处理计算、流计算和状态存储。在现实世界中，协调这些系统很容易出错，而且成本高昂，且难以自动化。

如果我们可以将事件存储配置为永久保存事件，那么就可以将批处理和流管道合并在一起。Pulsar 和 Kafka 都可以这么做，但是，Kafka 没有分层存储。这意味着所有事件都必须保存在 Kafka 节点的磁盘上。事件账本单调地增加，因此我们必须不断地添加存储空间。我们很少读取大多数历史事件，因此，我们昂贵的磁盘存储大部分都处于休眠状态。

另一方面，Apache Pulsar 具有内置的分层存储。Pulsar 将每个事件日志分写成段，并将不活动的段卸载（offloads）到 S3。这意味着只需对 Kafka 进行简单的配置更改，即可获得无限的、廉价的存储空间。我们无需不断地扩大集群的规模，就可以合并批处理和流管道。

我们可以将 Pulsar 配置为在主题达到特定大小时卸载事件，也可以手动运行。这使我们能够灵活地设置正确的卸载策略来平衡成本和速度。我们正在构建机器学习模型，以使我们的卸载策略适合每个主题的特定需求。

独立的计算和存储扩展

我们的事件数量和使用模式在一天之中和不同的用户之间变化很大。每个用户的不同使用模式会导致存储量或计算使用量的增加。幸运的是，Pulsar 将它的 Broker 从存储层分开了。

Pulsar 可以执行三种不同的操作：Tail write（尾部写入）、Tail read（尾部读取）和 Historical read（历史读取）。和 Kafka 一样，Pulsar 的写操作总是走到最后。对于 Pulsar 来说，一次写操作有三个步骤：首先，Broker 接收请求，然后 Broker 将其写入到 Bookkeeper，最后，它将请求缓存以供后续 Tail read。这意味着 Tail read 非常快，根本不会触及存储层。相比之下，Historical read 对存储层的影响就非常大。

对于 Kafka 和 Pulsar 来说，增加存储节点相对容易，但这是一项非常昂贵的操作。必须对数据进行洗牌（Shuffle）和复制，才能正确地平衡存储节点。在 Kafka 的情况下，Broker 和存储位于同一个节点上，因此任何扩展操作都很昂贵。与之相比，在 Pulsar，Broker 是无状态的，而且很容易进行扩展，成本也低廉。这意味着 Tail read 并不会带来严重的规模问题。我们可以根据当前 Historical read 和 Tail read 的使用模式来调整集群。

内置多租户

Pulsar 在构建的时候，就内置了多租户。在 Yahoo!，许多在不同地域分布的、从事不同产品工作的团队共享同一个 Pulsar 集群。该系统必须跟踪各种预算和各种服务级别协议。它有一个特性及，允许我们在同一个 Pulsar 集群上运行所有用户，同时保持性能、可靠性和安全性。

每个 Pulsar 主题都属于一个命名空间（Namespace），而每个命名空间都属于一个租户。每个 StreamSQL 账户都映射到一个租户。租户彼此安全隔离。用户不可能接触到其他用户的数据流。

从性能的角度来看，命名空间提供了关于隔离的其他有趣的动态。我们可以将用户的命名空间隔离到一组特定的 Broker 和存储节点。这限制了单个用户对整个系统的影响。同时，我们可以在 Broker 上设置自动减载，这样单个客户机中的峰值可以被更大的系统吸收。

积极响应的社区

Pulsar 社区的 Slack 频道真的太棒了！我的大多数问题几乎都能立即得到解答，还有一些聚会和 Pulsar 峰会，以及面对面的学习和网络课程。我们知道，哪怕在最糟糕的情况下，我们也可以联系相关人员，即使是最小众的问题也能得到帮助。社区给了我们继续前进的信心。