我在不同参数的循环中运行相同的模拟。每个模拟都使用一个 pandas DataFrame (data)，它只被读取，从不被修改。使用 ipyparallel(IPython 并行)，我可以在模拟开始之前将此 DataFrames 放入我 View 中每个引擎的全局变量空间中:

view['data'] = data

然后引擎可以访问 DataFrame 以获取在其上运行的所有模拟。复制数据(如果 pickled，data 为 40MB)的过程只有几秒钟。但是，如果模拟次数增加，内存使用量似乎会增长得非常大。我想这个共享数据正在被复制为每个任务 而不仅仅是为每个引擎。从客户端与引擎共享静态只读数据的最佳实践是什么？每个引擎复制一次是可以接受的，但理想情况下，每个主机只需要复制一次(我在 host1 上有 4 个引擎，在 host2 上有 8 个引擎)。

这是我的代码:

from ipyparallel import Client
import pandas as pd

rc = Client()
view = rc[:]  # use all engines
view.scatter('id', rc.ids, flatten=True)  # So we can track which engine performed what task

def do_simulation(tweaks):
    """ Run simulation with specified tweaks """
    #  Do sim stuff using the global data DataFrame
    return results, id, tweaks

if __name__ == '__main__':
    data = pd.read_sql("SELECT * FROM my_table", engine)
    threads = []  # store list of tweaks dicts
    for i in range(4):
        for j in range(5):
            for k in range(6):
                threads.append(dict(i=i, j=j, k=k)

    # Set up globals for each engine.  This is the read-only DataFrame
    view['data'] = data
    ar = view.map_async(do_simulation, threads)

    # Our async results should pop up over time.  Let's measure our progress:
    for idx, (results, id, tweaks) in enumerate(ar):
        print 'Progress: {}%: Simulation {} finished on engine {}'.format(100.0 * ar.progress / len(ar), idx, id)
        # Store results as a pickle for the future
        pfile = '{}_{}_{}.pickle'.format(tweaks['i'], tweaks['j'], tweaks['j'])
        # Save our results to a pickle file
        pd.to_pickle(results, out_file_path + pfile)

    print 'Total execution time: {} (serial time: {})'.format(ar.wall_time, ar.serial_time)

如果模拟计数很小(~50)，那么开始需要一段时间，但我开始看到进度打印语句。奇怪的是，多个任务将分配给同一个引擎，并且在为该引擎完成所有这些分配的任务之前我看不到响应。每次完成单个模拟任务时，我都希望看到 enumerate(ar) 的响应。

如果模拟计数很大(~1000)，启动需要很长时间，我看到所有引擎上的 CPU 节流，但直到很长时间(~40 分钟)才看到进度打印语句，当我确实看到了进展，似乎一大块(>100)任务进入了同一个引擎，并等待该引擎完成，然后再提供一些进展。当那个引擎完成时，我看到 ar 对象每 4 秒提供一次新响应——这可能是写入输出 pickle 文件的时间延迟。

最后，host1 还运行了 ipycontroller 任务，它的内存使用量疯狂上升(Python 任务显示使用 >6GB RAM，内核任务显示使用 3GB)。 host2 引擎并没有真正显示出太多的内存使用情况。什么会导致这种内存峰值？

最佳答案

我在几年前的代码中使用过这个逻辑，我开始使用 this .我的代码是这样的:

shared_dict = {
    # big dict with ~10k keys, each with a list of dicts
}

balancer = engines.load_balanced_view()

with engines[:].sync_imports(): # your 'view' variable 
    import pandas as pd
    import ujson as json

engines[:].push(shared_dict)

results = balancer.map(lambda i: (i, my_func(i)), id)
results_data = results.get()

If simulation counts are small (~50), then it takes a while to get started, but i start to see progress print statements. Strangely, multiple tasks will get assigned to the same engine and I don't see a response until all of those assigned tasks are completed for that engine. I would expect to see a response from enumerate(ar) every time a single simulation task completes.

在我的例子中，my_func() 是一个复杂的方法，我将大量日志消息写入一个文件，所以我有打印语句。

关于任务分配，因为我使用了 load_balanced_view()，所以我让图书馆自行其是，效果很好。

If simulation counts are large (~1000), it takes a long time to get started, i see the CPUs throttle up on all engines, but no progress print statements are seen until a long time (~40mins), and when I do see progress, it appears a large block (>100) of tasks went to same engine, and awaited completion from that one engine before providing some progress. When that one engine did complete, i saw the ar object provided new responses ever 4 secs - this may have been the time delay to write the output pickle files.

关于这么久了，我没有经历过，所以我什么也不能说。

我希望这能对您的问题有所启发。

PS:正如我在评论中所说，您可以尝试multiprocessing.Pool .我想我还没有尝试使用它来共享一个大的只读数据作为全局变量。我会试一试，因为 it seems to work .

关于python - 如何最好地在 ipyparallel 客户端和远程引擎之间共享静态数据？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/33722330/