1 个不稳定版本

0.13.0-dev.1	2023 年 12 月 15 日

1 在 #timely 中

每月下载量 2,298
用于 2 crate

MIT 许可证

685KB
10K SLoC

Timely Dataflow

Timely Dataflow 是一种低延迟的循环数据流计算模型，由论文 Naiad: a timely dataflow system 提出。本项目是 Rust 中对 timely dataflow 的扩展和更模块化的实现。

本项目类似于一个分布式数据并行计算引擎，可以将同一程序从笔记本电脑上的单线程扩展到跨计算机集群的分布式执行。主要目标是表达能力和高性能。假设您尚未使用 timely dataflow，则它可能比您当前使用的任何东西都要严格和快速。

请务必阅读 timely dataflow 的文档。它仍在进行中，但大多数方面都在改进。在 mdbook 格式的 mdbook 中有更多长文本，其中包含针对当前构建测试的示例。还有一个关于 timely dataflow 的系列博客文章（第一部分、第二部分、第三部分），以不同的方式介绍 timely dataflow，但请注意，那里的示例可能需要调整才能与当前代码兼容。

示例

要使用 timely dataflow，请将以下内容添加到项目 Cargo.toml 文件中的依赖项部分

[dependencies]
timely="*"

这将从 crates.io 引入 timely crate，这将允许您开始编写类似以下这样的 timely dataflow 程序（也可在 timely/examples/simple.rs 中找到）

extern crate timely;

use timely::dataflow::operators::*;

fn main() {
    timely::example(|scope| {
        (0..10).to_stream(scope)
               .inspect(|x| println!("seen: {:?}", x));
    });
}

您可以通过在 timely-dataflow 仓库的根目录中键入以下内容来运行此示例

% cargo run --example simple
Running `target/debug/examples/simple`
seen: 0
seen: 1
seen: 2
seen: 3
seen: 4
seen: 5
seen: 6
seen: 7
seen: 8
seen: 9

这是一个非常简单的示例（名称中已体现），仅展示如何编写数据流程序。

执行更多操作

对于更复杂的示例，请考虑类似的（但更明确的）examples/hello.rs，它分别创建和驱动数据流

extern crate timely;

use timely::dataflow::{InputHandle, ProbeHandle};
use timely::dataflow::operators::{Input, Exchange, Inspect, Probe};

fn main() {
    // initializes and runs a timely dataflow.
    timely::execute_from_args(std::env::args(), |worker| {

        let index = worker.index();
        let mut input = InputHandle::new();
        let mut probe = ProbeHandle::new();

        // create a new input, exchange data, and inspect its output
        worker.dataflow(|scope| {
            scope.input_from(&mut input)
                 .exchange(|x| *x)
                 .inspect(move |x| println!("worker {}:\thello {}", index, x))
                 .probe_with(&mut probe);
        });

        // introduce data and watch!
        for round in 0..10 {
            if index == 0 {
                input.send(round);
            }
            input.advance_to(round + 1);
            while probe.less_than(input.time()) {
                worker.step();
            }
        }
    }).unwrap();
}

此示例执行了许多操作，以展示 timely 可以为您做什么。

我们首先构建一个数据流图，创建一个输入流（使用input_from），其输出我们exchange来驱动记录在工作者之间（使用数据本身来指示路由到哪个工作者）。我们inspect数据并打印工作者索引以指示哪个工作者接收了哪些数据，然后probe结果，以便每个工作者可以看到给定一轮数据何时已全部处理。

然后我们通过重复引入数据轮次来驱动计算，其中round本身用作数据。在每一轮中，每个工作者引入相同的数据，然后重复执行数据流步骤，直到probe揭示所有工作者已处理该轮次的所有工作，此时计算继续。

对于两个工作者，输出看起来如下

% cargo run --example hello -- -w2
Running `target/debug/examples/hello -w2`
worker 0:   hello 0
worker 1:   hello 1
worker 0:   hello 2
worker 1:   hello 3
worker 0:   hello 4
worker 1:   hello 5
worker 0:   hello 6
worker 1:   hello 7
worker 0:   hello 8
worker 1:   hello 9

请注意，尽管工作者零引入了数据(0..10)，但每个元素都被路由到特定的工作者，正如我们预期的。

执行上面的hello.rs程序默认使用单个工作者线程。要在一个进程中使用多个线程，请使用-w或--workers选项，后跟您希望使用的线程数。（注意：simple.rs程序始终使用一个工作者线程；它使用timely::example，该程序忽略用户提供的输入）。要使用多个进程，您需要使用-h或--hostfile选项来指定一个文本文件，该文件的行是hostname:port条目，对应于您计划在其中生成进程的位置。您需要使用-n或--processes参数来指示您将生成多少个进程（主机文件的起始部分），并且每个进程都必须使用-p或--process参数来指示它们在这个数字中的索引。换句话说，您想要的hostfile看起来如下 % cat hostfile.txt host0:port host1:port host2:port host3:port ... 然后按如下方式启动进程 host0% cargo run -- -w 2 -h hostfile.txt -n 4 -p 0 host1% cargo run -- -w 2 -h hostfile.txt -n 4 -p 1 host2% cargo run -- -w 2 -h hostfile.txt -n 4 -p 2 host3% cargo run -- -w 2 -h hostfile.txt -n 4 -p 3 每个进程的工作者数应该相同。生态系统 Timely数据流旨在支持多个抽象级别，从最低级别的手动数据流组装到高级别的“声明式”抽象。目前有一些选项可以编写Timely数据流程序。理想情况下，随着感兴趣的人编写他们自己的层（或建立在他人的基础上），这个集合将随着时间的推移而扩展。 Timely数据流：Timely数据流包括几个原始操作符，包括标准操作符如map、filter和concat。它还包括更多异类操作符，如进入和退出循环（enter和leave），以及可以使用闭包提供实现的通用操作符（unary和binary）。微分数据流：在Timely数据流之上构建的高级语言，微分数据流包括group、join和iterate等操作符。其实现完全增量化，细节相当酷（如果神秘）。还有一些基于实时数据流的程序，包括流式最坏情况最优连接实现和PageRank实现，这两个都可以为编写实时数据流程序提供有用的示例。贡献如果您对与实时数据流合作或提供帮助感兴趣，那太好了！有一些工作类别对我们有帮助，也可能对您感兴趣。有几个广泛类别，然后是一系列不断变化的、各种复杂程度的问题。如果您想使用实时数据流编写程序，这对我们来说非常有趣。理想情况下，实时数据流应是一种用于非平凡类数据流计算的人性化方法。随着人们使用它并反馈他们的经验，我们了解他们发现的错误类别、用户体验痛点，以及其他我们事先甚至没有想象到的事情。了解实时数据流、尝试使用它并反馈是非常有帮助的！如果您喜欢编写小型示例程序或文档测试，实时数据流中有许多地方的示例相对稀疏，或者实际上并没有测试所展示的功能。这些通常很容易上手、完善并推进，而无需承担大量前期义务。如果您希望我们中有人详细解释某些内容，这也许是个好方法。如果您喜欢在实时数据流中亲自动手，问题跟踪器中有各种问题，涉及堆栈的不同级别。例如实时目前比必须的复制了更多的数据，这是为了最直接地满足Rust的所有权纪律。其中一些复制可以通过更仔细的资源管理来消除（例如，使用类似于bytes crate的一个Vec<u8>的共享区域）。这里并非所有内容都一目了然，因此也有机会进行一些设计工作。我们最近实施了一系列日志更改，但仍然有一个想要添加的功能列表还没有实现。如果您对通过探索记录其行为的底层基础设施来了解实时的工作方式感兴趣，这可能是一个不错的选择！此外，日志本身就是实时流，您甚至可以在实时上进行一些日志处理。哇... 关于将Rust所有权习惯语集成到实时数据流中有一个开放的问题。目前，实时流是可克隆对象的，当流被重用时，项目将被克隆。我们可以使其更明确，并要求调用一个.cloned()方法来获取拥有对象，就像迭代器需要它一样。同时，不获取所有权而使用流的引用应使您有机会查看经过的记录（而不需要所有权，也不需要克隆，如目前所做的那样）。这对于可能需要序列化数据且不能充分利用所有权的交换通道来说通常已经足够了。在调度及时数据流算子方面，有许多有趣的工作。当有机会调度许多算子时，我们可能会稍微思考一下，意识到其中一些没有工作要做，可以跳过。更好的是，我们可以维护一个需要执行工作的算子列表，对于没有工作的算子则不做任何事情。还有一些较大的工作主题，其解决方案并不明显，每个都有可能解决各种性能问题。速率控制输出目前，一元和二元算子的实现允许它们的闭包发送无界数量的输出。这可能导致不希望的资源耗尽，以及如果运行时需要分配大量新内存来缓冲批量发送的数据而无法消化它时，通常会导致性能下降。通常情况下，当产生大量数据时，如果有机会，它们最终会减少。在当前接口中，几乎没有可以做的事情。一个可能的变化是让input和notificator对象分别从一个输入消息或时间戳请求一个闭包到输出迭代器。这给了系统以它们认为适当的速度播放迭代器的机会。由于许多算子产生基于独立键的数据并行输出，构建这样的迭代器可能不会带来太大的负担。缓冲区管理及时通信层目前丢弃它通过交换通道移动的大多数缓冲区，因为它没有合理的方式来速率控制输出，也没有合理的方式来确定应该缓存多少个缓冲区。如果这两个问题中的任何一个得到解决，那么回收缓冲区以避免随机分配就变得有意义，特别是对于小批量数据。这些变化在dataflow-join三角形计算工作负载中具有大约10%-20%的性能影响。对不可序列化类型的支持通信层基于一个名为Content<T>的类型，它可以由类型化或二进制数据支持。因此，它需要支持该类型的可序列化，因为它需要为数据是二进制的情况提供逻辑，即使这种情况没有使用。看起来Stream类型应该可以扩展为在数据使用的存储类型上参数化，这样我们就可以表达某些类型不可序列化的事实，并且这是可以的。注意：差异数据流在它的operators/arrange.rs中演示了如何在用户级别做这件事，尽管有些简略（使用一个封装器来谎报它所传输类型的属性）。这将允许我们安全地传递Rc类型，只要我们使用Pipeline并行化合同。粗粒度与细粒度时间戳进度跟踪机制每个时间戳都涉及一些不平凡的开销。这意味着使用非常细粒度的时间戳，例如处理记录的纳秒，可能会淹没进度跟踪逻辑。相比之下，日志基础设施将纳秒降级为数据，成为日志有效负载的一部分，并用批次的最早时间戳近似事件批次。这在进度跟踪方面不那么准确，但性能更好。可能可以将这一点推广，使用户能够编写无需考虑时间戳粒度的程序，并且系统在可能的情况下自动粗化（本质上类似于boxcar-ing时间）。注意：差异数据流在它的collection.rs中演示了如何在用户级别做这件事。缺乏系统支持意味着用户最终需要指出粒度，这并不糟糕，但可能可以得到改进。也可能的是，将粒度控制留给用户，这使他们有更多的控制权来平衡延迟/吞吐量，这可能对系统来说是一件好事。

依赖关系 ~1.5–2.4MB ~47K SLoC abomonation abomonation_derive 0.5 crossbeam-channel futures-util getopts getopts 0.2.14 serde serde_derive timely-bytes-master timely-communication-master timely-container-master timely-logging-master dev rand+small_rng 其他功能 bincode