5

我正在考虑使用 Ray 来简单地实现数据的并行处理：

  • 有大量需要处理的数据项可以通过流/迭代器获得。每件物品的尺寸都很大
  • 应该在每个项目上运行一个函数，并将产生一个很大的结果
  • 处理后的数据应该在流中传递或存储在某种接收器中，该接收器只能在一段时间内接受一定数量的数据

我想知道这是否可以在 Ray 中完成。

目前我有以下基于pythons多处理库的简单实现：

  • 一个进程读取流并将项目传递给队列，该队列将在 k 个项目后阻塞（这样队列所需的内存不会超过某个限制）
  • 有几个工作进程将从输入队列中读取并处理项目。已处理的项目被传递到结果队列，该队列的大小再次受到限制
  • 另一个进程读取结果队列以传递项目

这样一来，一旦工作人员无法处理更多项目，队列就会阻塞，并且不会尝试将更多工作传递给工作人员。如果 sink 进程无法存储更多的项目，结果队列将阻塞，这反过来又会阻塞 worker，而 worker 又会阻塞输入队列，直到 writer 进程可以再次写入更多结果。

那么Ray 有抽象来做这样的事情吗？我如何确保只能将一定数量的工作传递给工作人员，我如何才能拥有像单进程输出函数这样的东西，并确保工作人员不能用如此多的结果淹没该函数，以至于内存/存储已用完？

4

2 回答 2

4

有一个适用于 Ray 的实验性流 API，您可能会发现它很有用：https ://github.com/ray-project/ray/tree/master/python/ray/experimental/streaming

它为流数据源、自定义运算符和接收器提供了基本结构。您还可以通过限制队列大小来为您的应用程序设置最大内存占用。

您能否分享一些有关您的应用程序的其他信息？

我们在谈论什么类型的数据？单个数据项有多大（以字节为单位）？

于 2019-05-08T23:14:52.613 回答
0

对于这个用例，我推荐 Ray 的并行迭代器。首先，您将创建一个生成器，该生成器从您的流式生成器中获取大对象（请参阅 参考资料ray.util.iter.from_iterators()）并对这些项目进行链式操作（请参阅参考资料.for_each()）。至关重要的是，中间对象（本身可能很大）一旦被链中的下一个函数消耗，就会从内存中逐出，从而防止内存不足。

最后，您可以控制队列上的执行，直到您的数据接收器准备好使用任何您想要的.take()方法。

于 2020-07-31T17:53:45.853 回答