伯克利 RISE Lab 开源 Confluo：吞吐量比 Kafka 高 4 到 10 倍

Confluo 是一个多数据流分析系统，可实现实时的分布式数据分析。Confluo 通过为多数据流的一些专门应用场景而精心设计的数据结构和针对端到端而优化的系统设计实现了高吞吐量并发写入、毫秒级在线查询和高效的即时查询。

我们很高兴将 Confluo 作为一个开源 C++ 项目，其中包括：

• Confluo 的数据结构库，支持高吞吐量日志摄入，以及各种在线（实时聚合、条件触发器执行等）和离线（即时过滤器、聚合等）的查询；

• Confluo 服务器实现，封装了数据结构，并提供 RPC 接口，以及 C++、 Java 和 Python 客户端库。

我们针对几种不同的应用场景对 Confluo 进行了评估，包括：

• 作为一个网络监控和诊断框架，Confluo 能够在单个核心上以线路速率（10Gbps 链路）执行数千个触发器和数十个过滤器。

• 作为一个时间序列数据库，与其他先进的时序数据库相比（如 CorfuDB、TimescaleDB 和 BTrDB），Confluo 的吞吐量提高了 2 之 20 倍，写入延迟降低了 2 至 10 倍，吞吐量提高了 1.5 至 5 倍，时间区间查询延迟降低了 5 至 20 倍。

• 作为一个 pub-sub 系统，Confluo 在发布订阅吞吐量方面是 Apache Kafka 的 4 至 10 倍。

Confluo 概览

很多现代应用程序，例如基于终端主机的网络监控、物联网和数字家庭一体化以及数据中心运营服务，它们的每台服务器每秒种都会捕获到数千万个数据点。这些数据被用于在线查询，实现可视化和监控，或者用于离线查询，进行故障分析和系统优化。要实现这些应用程序，需要一个实时监控和分析工具，能够支持高吞吐量数据摄取、低延迟的在线查询和低开销的离线查询。

虽然现在已有的数据结构可以支持高吞吐量数据摄取和丰富的在线和离线查询，但到目前为止，这两种数据结构仍然是互斥的。在从多个数据流摄取数据时，上述的查询需要更新多个数据结构——原始数据、聚合统计信息和物化视图。遗憾的是，用于支持这些查询的数据结构往往具有较高的更新开销，而且无法维持大多数应用程序所需的数据摄取速率。另一方面，可以维持高数据摄取速率的数据结构往往只支持非常简单的查询。

为了应对这一挑战，我们构建了 Confluo，一个同时实现了高吞吐量数据摄取和丰富的离线和在线查询的系统。

假设

Confluo 通过利用其目标应用程序语义来简化底层系统的假设，从而实现上述的目标。Confluo 的主要简化假设是：

• 应用程序数据流表现出一次性写入语义（即数据是追加写入的）；

• 监控和诊断应用程序使用固定大小的属性（例如，网络数据包中固定宽度的标头，分布式传感器网络中的 64 位时间戳和温度读数，数据中心操作指标中的浮点精度 CPU 和内存统计信息等）；

• 应用程序不需要事务性语义来进行并发操作，原子性语义就足够了。

Confluo API

Confluo 操作数据流，数据流由记录组成，记录使用了包含强类型属性集合的预定义模式（schema）。如上所述，Confluo 目前只支持固定大小的属性，包括原始数据类型，如二进制、整数或浮点数，或特定于域的类型，如 IP 地址、端口、传感器读数等。

Confluo 的模式是强类型属性的集合，语义类似于 JSON，例如，下面是一个带有五个属性的简单模式示例：

 复制代码

{
timestamp: LONG,
op_latency_ms: DOUBLE,
cpu_util: DOUBLE,
mem_avail: DOUBLE,
log_msg:STRING(100)
}

目前，Confluo 只支持具有固定模式的数据流，即数据流中的记录必须符合给定的模式。

为了加速即时离线查询，可以为模式中的属性添加索引。为了支持在线查询，Confluo 还采用一种匹配操作语言，其中包含三个主要元素：过滤器、聚合和触发器。

Confluo 过滤器是一种表达式，由任意有界宽度属性和关系运算符及布尔运算符组成（参见下表），用于标识与表达式匹配的记录。

Confluo 聚合（参见下表）用于计算与特定过滤器表达式相匹配的所有记录的属性的可计算函数。

Confluo 触发器是基于 Confluo 聚合计算得到的布尔条件（例如 <、>、= 等）。

Confluo 只支持为模式中具有固定大小的属性创建索引、过滤器、聚合和触发器。在添加好这些东西后，在新的数据记录到达时，它们都会被计算和更新。Confluo 目前不支持连接操作，因为在大多数监控和诊断应用程序中，这个操作并不常见。

实现

Confluo 使用了一种新的数据结构作为数据流的基本存储抽象：Atomic MultiLog，一组无锁并发日志，可用于存储原始数据、聚合统计信息和物化视图，并使用新的技术将整个集合作为单个原子操作进行更新。Atomic MultiLog 利用上述的应用程序工作负载假设来实现高吞吐量数据摄取和丰富的在线和离线查询。

Atomic MultiLogs 与数据库表的接口有点类似。为了存储来自不同流的数据，应用程序可以创建具有预定义模式的 Atomic MultiLog，并写入符合模式的数据流。然后，应用程序在 Atomic MultiLog 上创建索引、过滤器、聚合和触发器，为各种监控和诊断功能提供支持。

有关如何实现和使用 Confluo 的更多信息，请查看相关文档（ https://ucbrise.github.io/confluo/ ）。

性能

我们针对各种应用程序对 Confluo 进行了评估，包括网络监控和诊断、时间序列数据库和 pub-sub 消息系统。上图显示了 Confluo 在时间序列数据库应用程序中的性能表现，并将其与运行在配备了 18 个 CPU 内核和 60GB RAM 的 EC2 c4.8xlarge 实例上的 BTrDB、CorfuDB 和 TimescaleDB 进行了比较。我们使用了开放式 uPMU 数据集 的 5 亿个记录子集，这个数据集包含了安装在电网中的多个μPMU 的电压、电流和相位的读数，为期三个月。

我们发现，像 CorfuDB 和 TimescaleDB 这样的系统的性能比 BTrDB 和 Confluo 低 10 倍。但请注意，这不算是个缺点：CorfuDB 和 TimescaleDB 支持比 BTrDB 和 Confluo 更强的（事务性）语义。因此，根据所需语义的不同，任何一类系统对底层应用程序来说都可能是有用的。总而言之，与最先进的时间序列数据库相比，Confluo 的写入速度提高了 2 至 20 倍，写入延迟降低了 2 至 10 倍，时间区间过滤器的吞吐量提高了 1.5 至 5 倍，延迟降低了 5 至 20 倍。

网络监控和诊断 工具 的比较结果可以在我们即将发布的 NSDI 论文 中找到，而 pub-sub 系统的比较结果可以在 这里 找到。

限制

如前所述，Confluo 做了一些简化的假设，从而能够有效地实现各种在线和离线查询，同时支持每台服务器摄取数千万个数据点。因此，Confluo 只支持具有固定宽度的数据属性。此外，Confluo 目前只支持具有严格模式的流，不过我们也正在努力支持更灵活的模式。

展望未来

我们正在开发另外几个有趣的项目，以便让 Confluo 更具表现力并进一步提升效率。包括支持使用草图对数据流进行近似查询，支持基于数据流的 SQL 接口，以及通过文件合并和内存池来提高性能。要了解有关 Confluo 的更多信息，请访问我们的项目网站和 GitHub 存储库。

原文链接：

https://rise.cs.berkeley.edu/blog/confluo-millisecond-level-queries-on-large-scale-streaming-data/