如何在 Python 中并行化列表理解计算？

Question

列表推导式和映射计算都应该——至少在理论上——相对容易并行化：列表推导式中的每个计算都可以独立于所有其他元素的计算来完成。例如，在表达式中，

[ x*x for x in range(1000) ]

每个 x*x 计算可以（至少在理论上）并行完成。

我的问题是：是否有任何 Python 模块/Python 实现/Python 编程技巧来并行化列表理解计算（以便使用所有 16 / 32 / ... 核心或将计算分布在计算机网格上或在云上）？

Answer 1

正如 Ken 所说，它不能，但是使用 2.6 的 multiprocessing 模块，它很漂亮易于并行计算。

import multiprocessing

try:
    cpus = multiprocessing.cpu_count()
except NotImplementedError:
    cpus = 2   # arbitrary default


def square(n):
    return n * n

pool = multiprocessing.Pool(processes=cpus)
print(pool.map(square, range(1000)))

文档中还有一些示例，展示了如何使用管理器执行此操作，这也应该允许分布式计算。

Answer 2

对于共享内存并行，我推荐 joblib：

from joblib import delayed, Parallel

def square(x): return x*x
values = Parallel(n_jobs=NUM_CPUS)(delayed(square)(x) for x in range(1000))

Answer 3

关于列表理解的自动并行化

，恕我直言，如果没有附加信息（例如使用 OpenMP 中的指令提供的信息），或者将其限制为仅涉及内置类型的表达式，列表理解的自动并行化将不可能有效 /方法。

除非保证对每个列表项进行的处理没有副作用，否则如果不按顺序进行，结果可能会无效（或至少不同）。

# Artificial example
counter = 0

def g(x): # func with side-effect
    global counter
    counter = counter + 1
    return x + counter

vals = [g(i) for i in range(100)] # diff result when not done in order

还有任务分配的问题。问题空间应该如何分解？

如果每个元素的处理形成一个任务（〜任务场），那么当有许多元素每个都涉及琐碎计算时，管理任务的开销将淹没并行化的性能收益。

人们还可以采用数据分解方法，其中问题空间在可用进程之间平均划分。

事实上，列表理解也适用于生成器，这使得这有点棘手，但是如果预迭代的开销是可以接受的，这可能不是一个阻碍。当然，生成器也有可能产生副作用，如果过早地迭代后续项，则可能会改变结果。可能性很小，但有可能。

更大的问题是进程之间的负载不平衡。无法保证每个元素都会花费相同的时间来处理，因此静态分区的数据可能会导致一个进程完成大部分工作，而您的时间则被闲置。

将列表分解为更小的块并在每个子进程可用时处理它们是一个很好的折衷方案，但是，块大小的良好选择将取决于应用程序，因此如果没有用户的更多信息，则无法实现。

替代方案

正如其他几个答案中提到的，有许多方法和并行计算模块/框架可供选择取决于一项要求。

我只使用过 MPI（C 语言），没有使用 Python 进行并行处理的经验，因此我无法保证任何（尽管快速浏览后，
多处理，jug、pp 和 pyro 脱颖而出）。

如果要求尽可能接近列表理解，那么 jug 似乎是最接近的匹配。从 教程 中，跨多个实例分发任务可以非常简单

from jug.task import Task
from yourmodule import process_data
tasks = [Task(process_data,infile) for infile in glob('*.dat')]

：执行类似于 multiprocessing.Pool.map() 的操作，jug 可以使用不同的后端来同步进程并存储中间结果（redis、文件系统、内存中），这意味着进程可以跨越集群中的节点。

Answer 4

正如上面的答案所指出的，这实际上很难自动完成。那么我认为问题实际上是如何以最简单的方式做到这一点。理想情况下，解决方案不需要您知道“我有多少个核心”之类的信息。您可能想要的另一个属性是仍然能够在单个可读行中进行列表理解。

一些给定的答案似乎已经具有像这样的良好属性，但另一种选择是 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 `square` tasks automatically.
@ray.remote
def square(x):
    return x * x

# Create some parallel work using a list comprehension, then block until the
# results are ready with `ray.get`.
ray.get([square.remote(x) for x in range(1000)])

Answer 5

使用 futures.{Thread,Process}PoolExecutor.map(func, *iterables, timeout=None) 和 futures.as_completed(future_instances, timeout=None) 函数新的 3.2 concurrent.futures 包可能会有所帮助。

它还可以作为 2.6+ 向后移植 提供。

Answer 6

不，因为列表理解本身就是一种 C 优化宏。如果你把它拉出来并并行化它，那么它就不是一个列表理解，它只是一个很好的老式 MapReduce。

但您可以轻松地并行化您的示例。这里有一个关于将 MapReduce 与 Python 并行化库结合使用的很好的教程：

http: //mikecvet.wordpress.com/2010/07/02/parallel-mapreduce-in-python/

Answer 7

您可以使用asyncio。 （可以在[此处][1]找到文档）。它被用作多个 Python 异步框架的基础，这些框架提供高性能网络和 Web 服务器、数据库连接库、分布式任务队列等。此外，它还具有高级和低级 API 来解决任何类型的问题。

import asyncio

def background(f):
    def wrapped(*args, **kwargs):
        return asyncio.get_event_loop().run_in_executor(None, f, *args, **kwargs)

    return wrapped

@background
def your_function(argument):
    #code

现在，只要调用该函数，就会并行运行，而不会将主程序置于等待状态。您也可以使用它来并行化 for 循环。当调用 for 循环时，尽管循环是顺序的，但一旦解释器到达主程序，每次迭代都会与主程序并行运行。

对于您的特定情况，您可以执行以下操作：

import asyncio
import time


def background(f):
    def wrapped(*args, **kwargs):
        return asyncio.get_event_loop().run_in_executor(None, f, *args, **kwargs)                                   
    return wrapped


@background
def op(x):                                           # Do any operation you want
    time.sleep(1)
    print(f"function called for {x=}\n", end='')
    return x*x


loop = asyncio.get_event_loop()                      # Have a new event loop

looper = asyncio.gather(*[op(i) for i in range(20)]) # Run the loop; Doing for 20 for better demo
                             
results = loop.run_until_complete(looper)            # Wait until finish

print('List comprehension has finished and results are gathered!')      
print(results)

这会产生以下输出：

function called for x=5
function called for x=4
function called for x=2
function called for x=0
function called for x=6
function called for x=1
function called for x=7
function called for x=3
function called for x=8
function called for x=9
function called for x=10
function called for x=12
function called for x=11
function called for x=15
function called for x=13
function called for x=14
function called for x=16
function called for x=17
function called for x=18
function called for x=19
List comprehension has finished and results are gathered!
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81, 100, 121, 144, 169, 196, 225, 256, 289, 324, 361]

请注意，所有函数调用都是并行的，因此会打乱打印顺序，但原始顺序会保留在结果列表中。

Answer 8

AFAIK 不在列表理解范围内。

您当然可以使用传统的 for 循环和多处理/线程模块来做到这一点。

Answer 9

这里有一个 Python 并行包的完整列表：

http://wiki.python.org/moin/ParallelProcessing

我不确定是否有直接处理列表理解构造的拆分，但以非列表理解方式制定相同的问题应该很简单，可以轻松地分叉到多个不同的处理器。我不熟悉云计算并行化，但我在多核机器和集群上使用 mpi4py 取得了一些成功。您必须考虑的最大问题是通信开销是否会消除您从并行化问题中获得的任何收益。

编辑：以下内容可能也令人感兴趣：