我在考虑使用 Ray 来实现数据的并行处理:
- 有大量的数据需要通过流/迭代器进行处理。每个数据项都具有相当大的大小。
- 每个数据项都需要运行一个函数,该函数将产生相应的结果。
- 处理后的数据应通过流进行传递或存储在某种仅能在一定时间内接受一定数量数据的汇聚处。
我想知道是否可以用 Ray 实现这个过程。
目前,我已经基于 Python 多进程库创建了以下简单实现:
- 一个进程读取流并将其传递给一个队列,在 k 个数据项之后阻塞该队列(以使队列所需的内存不超过某个限制)。
- 有多个工作进程从输入队列中读取并处理数据项。处理后的数据传递到另一个大小有限的结果队列。
- 另一个进程从结果队列中读取数据项。
在此基础上,一旦工作进程无法再处理任何数据项,队列就会阻止,不会再向工作进程传递更多任务。 如果汇聚进程无法再存储更多数据项,则结果队列将被阻塞,这将进而阻塞工作进程,因此将阻塞输入队列,直到编写进程可以再次编写结果。
那么 Ray 是否有抽象层来完成类似的操作?我该如何确保只传递了一定量的任务给工作进程,并且如何实现单进程输出函数以确保工作进程不能通过太多的结果使内存/存储用尽?