Arroyo：基于Arrow和DataFusion的新SQL引擎


Arroyo 0.10 拥有一个使用 Apache Arrow 和 DataFusion 构建的全新 SQL 引擎。它更快、更小、更容易运行。

这篇文章将详细介绍 Arroyo 当前的实现以及为什么会发生变化，但简而言之：

  • 性能：Arrow 是一种内存中列格式，旨在利用现代 CPU 的矢量处理能力；与高性能计算内核相结合，我们可以实现最先进的流媒体性能，可与最好的批处理引擎竞争
  • 架构简单：今天，Arroyo 生成 Rust 代码，然后将其编译成执行数据处理的管道二进制文件。提前编译提供了良好的性能，但需要复杂的基础设施来编译管道。Arroyo 0.10 作为单个紧凑的二进制文件提供，可以通过多种方式进行部署。
  • 社区：Arroyo 正在迅速成为下一代数据堆栈的中心，通过采用它，Arroyo 可以与其他数据系统甚至其他语言（例如 Python）无缝交互。通过完全采用 DataFusion，我们能够利用（并贡献）新兴 Rust 数据生态系统的出色工作。

由于数据人员喜欢数字，以下是与 Arroyo 0.9 的一些比较：

  • 吞吐量：提高 3 倍
  • 管道启动：速度提高 20 倍
  • Docker 镜像大小：小 11 倍

截至今天，Arroyo 0.10 已作为开发者预览版提供。您可以通过运行 docker 容器开始：
docker run -it -p 8000:8000 ghcr.io/arroyosystems/arroyo-single:0.10.0-dev

那么我们是如何走到这一步的，为什么我们现在要做出这样的改变呢？让我们回顾一下 Arroyo 的历史，并介绍构建 SQL 引擎时的一些设计决策。

灵感由来
灵感来自于我在 Lyft 和 Splunk 领导 Apache Flink团队的经历。我亲眼目睹了开发 Flink 流处理管道的挑战以及操作的难度。在数据生态系统中，像 Redpanda 和 ScyllaDB 这样的项目成功地重新思考了现有的 Java 系统，用非托管语言实现了更简单、更高性能的实现，我认为 Flink 也有机会做同样的事情。

Arroyo 的最初原型以 Flink 作为直接灵感。我们的新系统将用更快的语言（Rust）编写，并将修复它的一些缺点，特别是在状态方面。

其他方面我们最初保持不变。例如，Flink 的核心 API 称为 Datastream API。它是一个用于直接定义 数据流图的Java API 。这是一个有向非循环图，数据消息在其边缘上流动，在实现查询逻辑各个部分（如过滤器、连接或窗口）的运算符之间流动。

在我们最初的 Arroyo 原型中，该图同样是通过 Rust API 直接定义的。

Stream::<()>::with_parallelism(1)
    .source(KafkaSource::new("localhost:9092", "events", HashMap::new()))
    .watermark(WatermarkType::Periodic {
        period: Duration::from_secs(1),
        max_lateness: Duration::ZERO,
    })
    .flat_map(|event| {
            event.split(" ").map(|w| (w.to_string(), 1)).collect()
    })
    .key_by(|(word, _)| word)
    .window(TumblingWindow::new(Duration::from_secs(1)))
    .sum_by(|(_, count)| count)
        .sink(KeyedConsoleSink::new());


但与 JVM 相比，Rust 这样的编译语言在这里提出了一些挑战。在 Flink 中，管道是用 Java 或 Scala 编写的，编译为 JVM 字节码，然后动态加载到 Flink 运行时中。但这种方法不太适合 Rust，它需要静态编译的二进制文件。

相反，我们将 Arroyo 运行时构建为一个库，它将由实际的管道代码调用。然后，所有内容都将被编译成静态二进制文件，该二进制文件将执行管道。

添加SQL
在 Lyft，SQL 是我们流媒体平台的潜在用户最需要的功能。虽然 Flink 有 SQL API，但用户测试表明它仍然太混乱，并且需要太多 Flink 专业知识，非流媒体专家才能掌握。

因此，我们从一开始就知道我们需要出色的 SQL 实现。我们希望了解 SQL 的数据工程师和科学家能够构建正确、可靠且高性能的流管道，而无需太多构建流系统的专业知识。

当我们开始构建 SQL 接口时，我们不想从头开始。因此，我们转向了 DataFusion，它既是一个完整的批处理 SQL 引擎，也是一个可组合的 SQL 原语库。我们决定只使用前端，它通过几个阶段获取用户提供的原始字符串 SQL：

  1. 解析，将 SQL 文本转换为抽象语法树 (AST)
  2. Planning，将 AST 转换为SQL 运算符的逻辑图，并具有它们之间的数据依赖关系
  3. 优化，将各种重写规则应用于图形以简化图形并使其执行效率更高

一旦我们有了优化的图，我们就把它转换成我们自己的数据流表示（物理图），我们将在运行时执行。

本质上，SQL 支持位于现有图形 API 之上。这实际上与 Flink SQL 的工作原理非常相似——SQL 由外部库（Apache Calcite）解析和规划，然后编译成 Flink Datastream 程序。

一旦确定了这个基本方法，我们就必须做出另外两个设计决策，这将决定我们未来的开发：如何表示 SQL 数据行以及如何实现 SQL 表达式。


在列上流式传输
在过去的十年中，几乎所有 OLAP（面向分析）引擎都采用了列表示5 。造成这种情况的原因有以下几个：

  • 通过将所有值存储在一列中，您可以获得更好的压缩比并更好地利用 CPU 缓存
  • 只需要读取查询中实际引用的列，减少磁盘和网络IO
  • 列式处理与现代 CPU 中的矢量功能非常契合，可提供 10 倍或更多的加速

然而，面向行的数据仍然是流引擎的标准。延迟（事件通过管道的速度）和吞吐量（给定数量的 CPU 可以处理多少个事件）之间存在一些固有的权衡。通过批处理数据，我们可以以延迟为代价获得更高的吞吐量。列式表示要求我们在看到性能改进之前将多个事件一起批处理（事实上，由于固定开销，行数较少的列式处理会慢得多）。

Flink 和 Arroyo 等流引擎在逻辑上一次对一个事件进行操作，并围绕有序处理提供重要保证。最初，它们是逐一对事件进行物理操作的。但批处理的好处不容忽视，最新版本的 Flink 确实 支持 SQL 运算符中的某些批处理[url=https://www.arroyo.dev/blog/why-arrow-and-datafusionuser-content-fn-5]6[/url]。

但我认为批处理在流处理中有意义的理由很简单： 在任何给定的批处理大小下，吞吐量越高，我们必须等待填充该批处理的时间就越少。例如，如果我们希望批次中至少有 100 条记录来克服固定成本，则等待接收 100 条记录所需的时间将取决于我们的吞吐量：

  • 如果每秒 10 个事件，则需要 1 秒
  • 1,000 时 — 0.01 秒 (100 毫秒)
  • 1,000,000 时 — 0.0001 (0.1ms)

或者从固定延迟的角度来看（例如，最多等待 10 毫秒）：
  • 在每秒 10 个事件时，我们的批量大小为 1
  • 1,000 — 100
  • 1,000,000 — 100,000

结论是：当我们的数据量非常低时，我们只需为小批量大小支付高额开销。但如果我们每秒只处理 10 或 100 个事件，那么无论如何，处理的总体成本都将非常小。在高数据量（每秒数万到数百万个事件）的情况下，我们可以鱼与熊掌兼得——通过批处理和列式数据实现高吞吐量，同时仍然保持较低的绝对延迟。

详细点击标题