python - dask:并行模型中的共享内存

Question

我已经阅读了 dask 文档、博客和 SO，但我仍然不是 100% 清楚如何去做。我的用例:

• 我有大约 10GB 的引用数据。一旦加载，它们就是只读的。通常我们将它们加载到 Dask/Pandas 数据框中
• 我需要这些引用数据每天处理(丰富、修改、转换)大约 500 个 mio 事件(多个文件)
• “流程”是一个包含大约 40 个任务的管道。执行顺序是相关的(依赖性)。
• 每个单独的任务都不复杂或耗时，主要是查找、扩充、映射等。
• 事件之间没有依赖关系。理论上，我可以通过单独的线程处理每个事件，将输出合并到一个文件中，这样就完成了。输出事件甚至不需要与输入事件的顺序相同。

总结:

• 我们可以大规模并行化事件处理
• 每个并行线程都需要相同的 10 GB(原始)引用数据
• 处理单个事件意味着对它们应用包含 40 个任务的序列/管道
• 每个单独的 Task 都不耗时(读取 ref-data 并修改事件)

可能的陷阱/问题:

• 花更多时间在序列化/反序列化上，而不是处理数据(我们在一些使用类似管道的方法的试验中确实遇到过这种情况)
• ref-data 被多次加载，每个(并行)进程加载一次
• 我最好在我的笔记本电脑上开发/测试它，但我没有足够的内存来加载引用数据。可能是解决方案是否会利用 memory_maps？

最有效的解决方案似乎是，如果我们只能将引用数据加载到内存中一次，使其对处理事件的多个其他进程以只读方式可用

通过在每台计算机中加载引用数据来扩展到多台计算机。将文件名推送到计算机以供执行。

知道如何实现吗？

非常感谢您的帮助

Answer 1

我也遇到过类似的问题，即运行令人尴尬的并行作业，这些作业都在同一个查找“引用”表(或并行进程的每个实例所需的任何大内存只读变量)中获取数据。只要当您处于遵循“写时复制”语义的环境中时(例如 linux)，将查找表放在全局范围内总是非常有效，如此处解释得很好: Shared-memory objects in multiprocessing

这是一个简单的并行工作流:

from multiprocessing import Pool

# Load your reference data, do that only once 
# here in the parent process
my_ref_lookup = load_ref_data(your_data_file)

def your_parallel_function(my_file_path):
    my_new_data = load_data(my_file_path)
    # process my_new_data with some lookup in my_ref_lookup 
    # which is known from the parent process. 

    processed_data = do_stuff(my_new_data)

    # you could here write something on disk
    # and/or return the processed_data

    return processed_data

with Pool(processes = 5) as Pool:
   list_of_result = Pool.map(your_parallel_function, your_list_of_file_paths)

此处 your_parallel_function 的执行将并行执行，例如5 个工作人员，一次在 your_list_of_file_paths 中获取 5 个文件，所有子进程都可以访问 my_ref_lookup 而无需复制它们。

在使用 Dask 和 bag 集合一段时间后，我从未发现与此类似或更简单的行为。在我尝试使用 Dask 时，在全局范围内以这种方式共享的只读变量最终被尽可能多的需要它的工作人员复制，这导致内存爆炸并导致我的内核崩溃。我从未在任何 Dask 文档中看到过这种情况。 Dask 文档中唯一与此相关的引用是关于避免全局状态:https://docs.dask.org/en/latest/delayed-best-practices.html#avoid-global-state但这显示了共享变量被延迟函数修改的情况，这与当前仅共享“只读”数据的问题不同。