AI前线导读：

伯克利RISE实验室又有新动作，最近开源了一个多数据流实时分布式分析系统Confluo。它可以作为网络监控和诊断框架，也可以作为时序数据库和发布订阅消息系统。作为时序数据库，它的性能比其他时序数据库高出数倍，而作为发布消息订阅系统，它的吞吐量比Kafka高出4到10倍。具体请见下文。

更多干货内容请关注微信公众号“AI前线”（ID：ai-front）

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/

创作场景

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