IPython 中的并行嵌套 for 循环

Question

我的 python 代码中有一个嵌套的 for 循环，看起来像这样：

results = []

for azimuth in azimuths:
    for zenith in zeniths:
        # Do various bits of stuff
        # Eventually get a result
        results.append(result)

我想在我的 4 核机器上并行化这个循环以加快速度。查看IPython并行编程文档（http://ipython.org/ipython-doc/dev/parallel/parallel_multiengine.html#quick-and-easy-parallelism）似乎有一种简单的方法来使用map 并行化迭代操作。

但是，要做到这一点，我需要将循环内的代码作为函数（这很容易做到），然后映射该函数。我遇到的问题是我无法获得一个数组来映射该函数。 itertools.product() 生成一个迭代器，我似乎无法使用 map 函数。

我尝试在这里使用地图是不是找错了树？有更好的方法吗？或者是否有某种方法可以使用 itertools.product，然后使用映射到结果的函数进行并行执行？

Answer 1

要并行化每个调用，您只需要获取每个参数的列表。您可以使用 itertools.product + zip 来获取：

allzeniths, allazimuths = zip(*itertools.product(zeniths, azimuths))

然后您可以使用 map：

amr = dview.map(f, allzeniths, allazimuths)

要更深入地了解这些步骤，这里有一个示例：

zeniths = range(1,4)
azimuths = range(6,8)

product = list(itertools.product(zeniths, azimuths))
# [(1, 6), (1, 7), (2, 6), (2, 7), (3, 6), (3, 7)]

所以我们有一个“对列表”，但我们真正想要的是每个参数的单个列表，即“列表对”。这正是稍微奇怪的 zip(*product) 语法给我们带来的结果：

allzeniths, allazimuths = zip(*itertools.product(zeniths, azimuths))

print allzeniths
# (1, 1, 2, 2, 3, 3)
print allazimuths
# (6, 7, 6, 7, 6, 7)

现在我们只需将函数映射到这两个列表上，以并行化嵌套的 for 循环：

def f(z,a):
    return z*a

view.map(f, allzeniths, allazimuths)

并且只有两个没有什么特别的 -此方法应该扩展到任意数量的嵌套循环。

Answer 2

我假设您使用的是 IPython 0.11 或更高版本。首先定义一个简单的函数。

def foo(azimuth, zenith):
    # Do various bits of stuff
    # Eventually get a result
    return result

然后使用 IPython 的精细并行套件来并行化您的问题。首先通过在终端窗口中启动集群来启动连接有 5 个引擎 (#CPUs + 1) 的控制器（如果您安装了 IPython 0.11 或更高版本，则应该存在此程序）：

ipcluster start -n 5

在您的脚本中连接到控制器并传输所有任务。控制器会处理一切。

from IPython.parallel import Client

c = Client()   # here is where the client establishes the connection
lv = c.load_balanced_view()   # this object represents the engines (workers)

tasks = []
for azimuth in azimuths:
    for zenith in zeniths:
        tasks.append(lv.apply(foo, azimuth, zenith))

result = [task.get() for task in tasks]  # blocks until all results are back

Answer 3

我不太熟悉 IPython，但一个简单的解决方案似乎是仅并行化外循环。

def f(azimuth):
    results = []
    for zenith in zeniths:
        #compute result
        results.append(result)
    return results

allresults = map(f, azimuths)

Answer 4

如果您确实想并行运行代码，请使用concurrent.futures

import itertools
import concurrent.futures

def _work_horse(azimuth, zenith):
    #DO HEAVY WORK HERE
    return result

futures = []
with concurrent.futures.ProcessPoolExecutor() as executor:
    for arg_set in itertools.product(zeniths, azimuths):
        futures.append(executor.submit(_work_horse, *arg_set))
executor.shutdown(wait=True)

# Will time out after one hour.
results = [future.result(3600) for future in futures]

Answer 5

如果您想保留循环的结构，可以尝试使用 Ray (docs) ，这是一个用于编写并行和分布式Python的框架。一个要求是您必须将可以并行化的工作分离到其自己的函数中。

您可以像这样导入 Ray：

import ray

# Start Ray. This creates some processes that can do work in parallel.
ray.init()

然后，您的脚本将如下所示：

# Add this line to signify that the function can be run in parallel (as a
# "task"). Ray will load-balance different `work` tasks automatically.
@ray.remote
def work(azimuth, zenith):
  # Do various bits of stuff
  # Eventually get a result
  return result

results = []

for azimuth in azimuths:
    for zenith in zeniths:
        # Store a future, which represents the future result of `work`.
        results.append(work.remote(azimuth, zenith))

# Block until the results are ready with `ray.get`.
results = ray.get(results)