流数据并行处理性能比较：Kafka vs Pulsar vs Praveg

1、引言

流式应用程序通常从各种各样的来源 (例如，传感器、用户、服务器) 并发地采集数据，并形成一个事件流 (stream of events)。使用单个流来捕获由多个数据源生成的并行数据流可以使得应用程序能够更好地理解数据，甚至更有效地处理数据。例如，将来自一组传感器的数据输入到单一数据流中，就可以使得应用程序通过引用单一数据流来分析所有这类传感器数据。当这些单个的流可以以高并行度读取时，应用程序就能自行决定如何映射自身的抽象设计到这些流进行数据读取，而不是被人为的基础设施限制而决定。

 

并行化在处理流数据时也很重要。当应用程序分析流中的数据时，它们通常依赖并行处理来降低延迟和提高吞吐量。为了在读取流式数据时支持并行性，流存储系统允许在数据写入时，根据事件负载进行分区。这通常基于路由键 (routing keys) 的支持。通过分区，应用程序可以保留以应用本身概念（如标识符）的顺序。在每个分区内，数据是有序的。在 Pravega 中，Stream 的并行单位被叫做 segment，而在基于 topic 的系统中 (如 Apache Kafka 和 Apache Pulsar)，它被称为 partitions。Pravega 的 stream 可以自动根据负载的变化改变 segment 的数量。

 

在本文中，我们关注的是 Pravega 在多个写客户端同时写入到一个多 segment 的 stream 时的表现。我们需要特别关注下数据的添加路径 (append path)，因为它对于有效地读取数据流至关重要。我们已经在之前的博客中分析了在只有一个写入端和最多 16 个 segment 的基本 IO 性能。这一次，我们使用高度并行的负载，每个流最多有 100 个写入端和 5000 个 segment。这样的设置参考了当今云原生应用程序的需求，例如对于高度并行的工作负载，它们对于扩展和维持高性能的需求。我们将 Pravega 与 Kafka 和 Pulsar 进行比较，以了解这些系统因为不同的设计而带来的影响。我们注意到的两个关键点是:

 

在最多有 100 个写入端和 5000 个 segment，数据流量在 250MBps 时，Pravega 可以在所有情况下都维持 250MBps 的速率。而 Kafka 在 5000 个 partition 时只有不到 100MBps。Pulsar 则在大多数情况下会直接崩溃。

 

Pravega 可以保证在 95% 中位数时，延迟在 10 毫秒以下，而 Kafka 的延迟却高达几十毫秒。对于 Pulsar，它有高达几秒的延迟，并且是在我们仅仅能成功的那次。

 

为了公平起见，我们已经测试了 Pulsar 的其他配置，以了解在哪些条件下它表现出良好的性能结果。因此，我们另外展示了一个对 Pulsar 更有利的配置，并且不会导致它经常崩溃。但这个配置对于我们的测试系统来说没有那么大的挑战，如我们在下面进一步解释的那样。

 

在下面的章节中，我们将解释是什么能够让 Pravega 在这种情况下表现得更好，并详细介绍我们的环境设置、实验过程和结果。

 

2、为什么 Pravega 性能更好？

我们介绍一些关于 Pravega 添加路径 (append path) 的设计特点，这些特点对于理解结果很重要。我们还讨论了一些有关设计的权衡，并阐述了我们为什么在 Pravega 上选择这种。

 

Pravega 的添加路径 (append path)

Pravega 的添加路径 (append path) 包括三个相关部分:

添加数据的客户端

Segment Store，用以接收数据添加的请求，记录其日志并持久化存储

持久化的日志存储，由 Apache BookKeeper 实现

 

下图阐释了 Pravega 的添加路径 (append path):

 

Pravega 的添加路径 (append path)

 

客户端添加由程序源生成的数据，并尽可能对这些数据进行批处理。在服务端收集客户端的批处理数据，这样做的好处时可以避免缓冲数据，但要注意是由客户端来控制批处理何时开始和结束。客户端使用了一种批处理跟踪的启发式，这个算法通过输入速率和响应反馈来估计批处理的大小。有了这样的估计，客户端就可以决定何时关闭批处理 (代码在 AppendBatchSizeTrackerImpl.)。

 

由于客户端批处理的大小最终取决于应用程序源可以生成多少数据，因此很有可能单个客户端自己无法生成足够大的批处理。因此，当有多个写入端时，我们有机会聚合来自多个客户端的批处理，以形成更大的批处理。实际上，这就是 segment store 的关键作用之一: 在将数据写入持久性日志之前对其进行聚合，其中的细节读者有兴趣可以延申阅读本篇博客。

 

segment store 在一个称为 segment 容器 (segment container) 的组件中执行这第二级批处理。一个 segment store 可以同时运行多个 segment 容器，并且每个容器都有自己的持久化日志 (durable log) 并追加到其中。这些 segment container 在单个 segment store 实例中可以并行地进行数据的 append 写入。

 

每个 segment 容器负责在 Pravega 集群中所有 stream 中一部分 segment 上的操作。在全局上，集群会协调哪些 segment 容器对应哪些 segment store 实例。当增加 segment store 实例时 (例如扩大系统规模) 或减少 segment store 实例时 (如缩减系统规模或局部宕机)，segment 容器集合将跨现有 segment store 实例重新平衡。

 

持久性日志目前是通过 Apache BookKeeper ledgers 实现的。Bookie(BookKeeper 的存储服务器) 将数据添加请求的日志记录到 ledgers 中，并在将数据添加加到 journals 之前执行另一层合并。这第三层的合并又是一个批量处理来自不同 segment 容器数据的机会。在我们用于 Pravega 的配置中，为了保证持久性，bookie 只对将数据写入 journal 的分 segment 容器做出响应。BookKeeper 还维护其他数据结构，但它们与本文的讨论无关。

 

低延迟、高吞吐量和持久性

低延迟对于许多流式应用程序是至关重要的。这些应用要求数据在生成后不久就可以进行处理。高吞吐量同样适用于需要从许多来源获取大量数据的应用程序。如果系统不能够维持高吞吐量的输入，应用程序在数据峰值时可能会面临不得不需要局部减载的危险。最后，在分析并处理这些流时，数据的丢失可能导致不正确的结果，因此，持久性对于企业应用程序也是至关重要。

 

然而，在一套系统里同时实现这三个特性是具有挑战性的。存储设备通常通过更大块的写入来提高吞吐量，迫使系统缓存更多的数据并且牺牲了延迟。如果系统偏向更低的延迟，那么每次写入时的数据就会减少，从而降低了吞吐量。当然，如果我们不在数据刷新到磁盘后确认，那么我们就可以不关注延迟和吞吐量之间的权衡，但是这种选择牺牲了持久性。

 

在我们评估的所有三个系统中，Pravega 在这三个方面总体上提供了较好的结果。与 Kafka 和 Pulsar 相比，它在保持高吞吐量和低延迟的同时保证了持久性。Kafka 则在这三个方面做出了不同的选择。与其他两种配置相比，默认情况下它可以获得更高的吞吐量和更低的延迟，因为它不会等待磁盘的写入成功，但这种选择牺牲了持久性。Pulsar 能够像 Pravega 一样保证持久性，因为它建立在 BookKeeper 的基础上 ; 尽管如此，它似乎并没有实现一个写路径 (write path)，能够保证在多写入客户端和多 partition 的情况下依然足够高效，正如之后与其他两个系统相比的实验结果所展示的那样。

 

3、评测与配置一览

我们在 AWS 上进行了实验。我们在 Pravega 中使用的方法非常接近我们在前一篇博文中描述的方法，可以参考那篇博文了解更多细节。与我们之前的博客不同的是，公平起见，对于 Pulsar，类似于对 Kafka 做的，我们使用了 StreamNative 之前的性能分析博客中使用的相同的 OpenMessaging 测试工具代码。在下面，我们为这篇博文提供了主要的配置设置:

 

在我们性能测试中使用的主要设置

 

在我们之前的博客中，我们对 Pravega、 Apache Pulsar 和 Apache Kafka 使用了相同的副本方案: 3 个副本，每次收到 2 个确认进行写入。对于数据持久性，Pravega 和 Pulsar 在默认情况下保证每次写入的数据持久性，并且我们保留这种行为。对于 Kafka，我们测试了两种配置: 1) 默认配置 no flush，这种配置下数据不会显式地刷到磁盘中，那么在有些相关的故障发生时可能导致数据丢失 ; 2) 磁盘刷新配置，其通过每次写入刷入磁盘的方式保证数据的持久性。日志在所有的系统都写入了一个 NVMe 硬盘，这样我们可以理解这三个系统如何在并发度上升的时候使用它。

 

我们部署的硬件参数与之前的博客不同。这次的实验使用了 AWS 更大的服务器实例 (对于 Pulsar Broker 和 Pravega segment store 我们使用了 i3.16xlarge 实例)。更改实例的原因是，我们观察到所有这些系统在有更多的 partition/segment 时，会使用更多的 CPU。通过增加 CPU 资源，我们才能保证这些系统不会被让 CPU 成为性能瓶颈。我们还使用多个测试虚拟机。通过模拟一个分布式并行数据源 (Pravega、 Kafka 和 Pulsar 的 terraform 配置文件) ，这个部署符合我们在高负载下测试这些系统的目标。还要注意的是，在这个评估中，Pravega 是将数据转移到长期存储（AWS EFS）的系统。

 

我们使用 OpenMessaging 基准测试来运行我们的实验 (请参阅这里的部署说明并在 Pravega 实验中使用此版本)。我们将输入数据速率固定在 250MBps(每个事件 1KB)，而不是探索和评估系统的较大吞吐量。我们的目标是在客户端数量和 segment/partition 的数量改变时，比较的不同系统的行为。这样的设置使得实验可以保证数据注入的固定以减小其他因素对性能测试的影响。为了完整起见，我们仍然在博客的末尾讨论了这些系统的较大吞吐量。

 

测试中的生产者和消费者线程分布在很多虚拟机上 (详情请参见上表中的 Producers/Consumers 行)。每个生产线程和消费线程都使用一个专用的 Kafka、 Pulsar 或 Pravega 客户端实例。基准测试的生产者线程使用 Kafka 和 Pulsar 中的 producer 或 Pravega 中的 writer ，而基准测试的消费者线程使用 Kafka 和 Pulsar 中的 consumer 或 Pravega 中的 reader。

 

在 Pravega 的具体案例中，写入端使用了连接池技术: 这个特性允许应用程序使用一个公共的网络连接池 (默认情况下每个 segment store10 个连接) 来处理大量的 segment 连接。

 

数据输入和并行性

我们要评估的第一个方面是增加 segment 和客户端数量对吞吐量的影响。下面的图表显示了 Pravega、 Kafka 和 Pulsar 的吞吐量，包括不同数量的 segment 和生产者。每一条线对应于在同一个工作负载下，向单一一个 stream/topic 追加不同数量的生产者。对于 Kafka 和 Pulsar，我们也绘制了其他配置下的图线。这些配置以牺牲功能性（持久性或者是 no key）为代价，为这两个系统提供了更好的结果。

 

 

该实验可以通过 P3 测试程序复现，它使用以下的工作负载和配置文件作为输入。他们分别是 Pravega（工作负载，配置），Kafka（工作负载，配置）和 Pulsar（工作负载，配置）。这些系统的原始基准测试输出可在此处获得：Pravega，Kafka(不 flush)，Kafka(flush)，Pulsar，Pulsar(有利配置)。

 

通过研究上面的实验图表，我们观察到以下关于吞吐量和并行性的关系：

Pravega 是这些系统中可以在 250MBps 数据流，5000 个 segment 和 100 个生产者的负载下稳定工作的。这表明 Pravega 添加路径 (append path) 的设计可以有效地处理高并行下的工作负载，特别是当许多写入端的小数据的追加写在 segment 容器中进行批量化处理的设计。

 

在我们增加 partitions 数量后，Kafka 的吞吐量降低了。更多的写入者不会线性地增加吞吐量，而是存在一个性能上限。换句话来说，10 个和 50 个生产者在吞吐量上会有显著的区别，但是 50 个和 100 个生产者的差距对于吞吐量的影响不大。这个结果证明了很多 Kafka 用户共同的担忧，即增加 partition 数量后，Kafka 的性能会下降。

 

值得一提的是，当我们在 Kafka 的配置中开启了 flush 强制刷新到磁盘以确保数据持久性，吞吐量有着显著的下降 (例如对于 100 个生产者和 500 个分区时，有 80% 的下降)。虽然强制刷新到磁盘会对系统造成一定的性能损失，但是这个实验表明了在超过十个分区且保证持久性的情况下，吞吐量会有非常明显的损失。

 

Pulsar 在我们试验过的大多数配置情况下都会崩溃。为了了解 Pulsar 稳定性问题的根本原因，我们换了一个更有利的配置:

等待所有来自 Bookies 的确认请求，这样可以解决 out-of-memory 内存不足的错误 (更多细节见这个 issue)

不使用路由键来写入数据 (这样会牺牲数据的有序性并且降低实际的写入并行性)。如果启动的话，我们会在 Pulsar 的 broker 中发现关于 Bookkeeper 的 DBLedgerStorage 不能够跟上写入的速度的错误。

 

通过这种配置，Pulsar 可以得到比基本情景 (如 10 个生产者) 更好的结果。然而，当实验中有大量的生产者和分区时，它仍然显示出性能下降和最终的不稳定性。注意，在写操作中不使用路由键是 Pulsar 性能提升的主要原因。在内部，Pulsar 客户端通过创建更大的批处理并以 round-robin 的方式使用 segment 来优化没有键的情况 (参见文档)。

 

总之，Kafka 和 Pulsar 在增加分区和生产者数量时都会显著降低性能。需要高度并行性的应用程序可能无法满足所需的性能要求，或者不得不在这个问题上投入更多资源。Pravega 是经过测试下，一个可以在大量的生产者和 segment 的规模下依然保持持续高吞吐量的系统，同时在此配置下数据的持久性也能得到保证。

 

关注写入延迟

在写入流数据时，延迟也和吞吐量一样重要。接下来，我们展示了 95% 中位数时写入延迟和 segment/ 生产者数的关系。请注意，在本节中，我们展现了所有系统的延迟数据，而不考虑它们是否达到了要求的高吞吐量。

 

 

 

通过研究上面的实验图表，我们重点总结出了以下结论:

Pravega 在 95% 中位数的情况下提供<10 毫秒的延迟，而 Kafka，即使修改配置使其不刷新到磁盘，也有更高的延迟。回想一下上一节，Pravega 在有很多 segment 和生产者的时候也能达到目标的高吞吐量，而 Kafka 没有。

 

对于 5000 个 segment，与 100 个生产者的情况相比，Pravega 在 10 个生产者下能获得更低的延迟。这样的结果是由于批量化处理的设计。随着生产者的增加，Pravega 的添加路径 (append path) 使得单个生产者的每一批次的变小，这导致了数据在服务器端排队和需要更多的计算量。

 

Kafka 在保证数据持久性 (即打开 flush 开关) 的模式下，延迟比默认配置更高了 (95% 中位数的延迟在 100 个生产者和 500 个 segment 的情况下达到了 13.6 倍的延迟)。在高并行的需求下，应用程序可能不得不在数据持久性和高并行的性能下二选一。

 

基本配置下的 Pulsar 只能在 10 个 partition + 10 个生产者的情况下保持低延迟。任何提升 partition 数量或者写客户端数量都会导致系统不稳定。

当使用更有利的配置时，Pulsar 在 10 个生产者的情况下可以获得<10 毫秒的延迟。对于 100 个生产者，延迟会随着 partition 数量的增加而迅速上升。

与 Kafka 相比，Pravega 拥有更低的延迟，甚至对于默认的不等待磁盘返回写入确认的 Kafka 配置来说也是如此。对于 Pulsar，系统的不稳定性不允许我们对建立的配置进行干净的比较。对于更有利的 no key + 等待所有 bookie 返回的配置下，Pulsar 能在 10 个生产者下保持了<10 毫秒延迟，但是对于 100 个生产者的话延迟会增加的很快。

 

关于较大吞吐量的附注

虽然我们在上面的实验中使用了一个固定的目标速率，但是我们还想了解这些系统在我们的场景中能够达到的较大吞吐量。为了缩小分析范围，我们固定使用 10 个生产者，对比了在 10 和 500 个 segment/partition 时的情况。

 

这些测试的原始测试结果可以在这里看到。

 

从基准测试的角度来看，Pravega 可以在 10 和 500 个 segment 下达到 720MBps 的较大吞吐量，换算在磁盘级别则大约是 780MBps。这个差异是由于 Pravega 和 BookKeeper 添加了额外的元数据开销 (例如 Pravega 的 segment 属性)。这个成绩非常接近同步写入磁盘的较大速度（约 803MBps）。实验如下:

[ec2-user@ip-10-0-0-100 ~]$ sudo dd if=/dev/zero of=/mnt/journal/test-500K bs=500K count=100000 oflag=direct100000+0 records in100000+0 records out51200000000 bytes (51 GB) copied, 63.7856 s, 803 MB/s

 

对于自定义配置的 Pulsar，我们可以在基准测试上可靠地获得近 400MBps 的吞吐量。我们还测试了将客户端的批处理时间增加到 10 毫秒，这使得吞吐量略有提高 (515MBps)。尽管如此，我们注意到这远远不是磁盘写入的较大速率，我们怀疑这是由于使用了路由键，因为它减少了 Pulsar 客户端批处理的机会。更糟糕的是，随着分区数量的增加，Pulsar 的吞吐量很快受到了限制。这个结果表明，主要依赖客户端来聚合数据成批具有很大的局限性。

 

对于有 10 个 partition 的情况，我们观察到，当 Kafka 保证持久性 (“flush” 模式) 时，它可以在等待写返回时达到 700MBps 和不等待写返回时达到 900MBps。请注意，这只发生在有 10 个 partition 的情况下，因为当有 500 个 partition 时，吞吐量分别降至 22MBps 和 140MBps。

 

为了获得更深入的了解，我们在执行实验时使用 iostat 对服务器端实例进行了检测。根据从 iostat 每秒收集的信息，Kafka 使用 no flush 的设置允许操作系统缓冲更多的数据，再写到硬盘，从而导致更高的吞吐量。下面的图表从操作系统的角度展示了 Kafka 写入的行为: 写往往是 250KB 的大小，或者因为缓冲根本没有写入 (0 大小)。相反，Pravega 显示了一个更小但是一致的写大小，因为每一次写都被刷新到硬盘，并且 Pravega 添加路径 (append path) 定义了它们的大小。注意，即使牺牲持久性，Kafka 也只能在更少的分区上达到这样的吞吐率。

 

在不同 segment 或写入者的情况下，iostat 监测的平均磁盘写入的大小的累积分布函数

 

4、总结

随着越来越多的需要读写并行化的实际使用情况，流存储有效地、高效地适配这些工作负载变得至关重要。许多这样的应用程序是云原生的，并且它们需要有效地伸缩和并行化这些工作负载的能力。

 

这篇博客展示了，Pravega 能够在保持低延迟和数据持久性的同时，维持数千个 segment 和数十个并发写入者的高吞吐量。同时提供高吞吐量、低延迟和数据持久性是一个具有挑战性的问题。我们比较的 Kafka 和 Pulsar 这些消息传递系统在同样的测试数据集上还不够优秀。Pravega 的添加路径 (append path) 包括了多个批处理步骤，在保证数据持久性的同时，还能应对高并行度工作负载的挑战。今后还会有更多的性能测试博客。请保持关注。