1 个不稳定版本

0.13.0-dev.1	2023年12月15日

3 在 #timely

2,317 每月下载量
在 4 个crate中使用了（2个直接使用）

MIT 许可证

12KB
170 行

Timely Dataflow

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

本项目类似于一种分布式数据并行计算引擎，可以将相同的程序从笔记本电脑上的单个线程扩展到集群中的分布式执行。主要目标是表达能力和高性能。假设您尚未使用timely dataflow，那么它可能比您目前正在使用的任何东西都更加表达丰富且速度更快。

请务必阅读timely dataflow文档。它仍在进行中，但总体上在改进。在mdbook格式中还有更多长篇文本，带有与当前构建测试的示例。还有一个系列博客文章（第一部分、第二部分、第三部分），以不同的方式介绍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引入了数据 (0..10)，但每个元素都按照我们的意图路由到了特定的工作者。

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

无运行时依赖