【问题标题】:Parallel Processing in pythonpython中的并行处理
【发布时间】:2010-10-01 18:59:55
【问题描述】:

在 python 2.7 中进行并行处理的简单代码是什么?我在网上找到的所有示例都令人费解,并且包含不必要的代码。

我将如何做一个简单的蛮力整数分解程序,我可以在每个核心 (4) 上分解 1 个整数?我真正的程序可能只需要 2 个内核,并且需要共享信息。

我知道存在 parallel-python 和其他库,但我想将使用的库数量保持在最低限度,因此我想使用 thread 和/或 multiprocessing 库,因为它们带有 python

【问题讨论】:

标签: python parallel-processing


【解决方案1】:

在 python 中开始并行处理的一个很好的简单方法就是多处理中的池映射——它类似于通常的 python 映射,但单个函数调用分布在不同数量的进程中。

因式分解就是一个很好的例子 - 您可以强力检查分布在所有可用任务上的所有部门:

from multiprocessing import Pool
import numpy

numToFactor = 976

def isFactor(x):
    result = None
    div = (numToFactor / x)
    if div*x == numToFactor:
        result = (x,div)
    return result

if __name__ == '__main__':
    pool = Pool(processes=4)
    possibleFactors = range(1,int(numpy.floor(numpy.sqrt(numToFactor)))+1)
    print 'Checking ', possibleFactors
    result = pool.map(isFactor, possibleFactors)
    cleaned = [x for x in result if not x is None]
    print 'Factors are', cleaned

这给了我

Checking  [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31]
Factors are [(1, 976), (2, 488), (4, 244), (8, 122), (16, 61)]

【讨论】:

  • 我应该补充一点,上述方法有效,但可能不会执行惊人的并行性能壮举,因为您调用的开销(并行映射 + 函数调用)全部用于计算少量工作(一点点整数运算)。我将把它作为练习留给读者思考如何将开销分摊到更多的部门——例如,如何更改上面的代码,以便为多个部门调用一次“isFactor”。
  • 示例代码给出错误AttributeError: Can't get attribute 'isFactor' on <module '__main__' (built-in)>
  • Python 3 注意:isFactor(x)中的numToFactor / x,替换成整数除法(//)
【解决方案2】:

mincemeat 是我发现的最简单的 map/reduce 实现。此外,它对依赖项非常轻——它是一个文件,并且使用标准库完成所有操作。

【讨论】:

  • 有趣...我会研究一下
  • 这不是我真正想要的
  • @calccrypto 为什么不呢?了解肉馅为何不完美可能有助于其他人找到更好的解决方案。
  • 它更多地用于数据库和服务器之类的东西(不止一台计算机)。我只是想一次运行多个功能
【解决方案3】:

我同意,如果您想留在标准库中,使用 multiprocessing 中的 Pool 可能是最好的方法。如果您对做其他类型的并行处理感兴趣,但没有学习任何新东西(即仍然使用与multiprocessing 相同的界面),那么您可以尝试pathos,它提供了多种形式的并行映射,并且几乎具有与multiprocessing 相同的界面。

Python 2.7.6 (default, Nov 12 2013, 13:26:39) 
[GCC 4.2.1 Compatible Apple Clang 4.1 ((tags/Apple/clang-421.11.66))] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> import numpy
>>> numToFactor = 976
>>> def isFactor(x):
...   result = None
...   div = (numToFactor / x)
...   if div*x == numToFactor:
...     result = (x,div)
...   return result
... 
>>> from pathos.multiprocessing import ProcessingPool as MPool
>>> p = MPool(4)
>>> possible = range(1,int(numpy.floor(numpy.sqrt(numToFactor)))+1)
>>> # standard blocking map
>>> result = [x for x in p.map(isFactor, possible) if x is not None]
>>> print result
[(1, 976), (2, 488), (4, 244), (8, 122), (16, 61)]
>>>
>>> # asynchronous map (there's also iterative maps too)
>>> obj = p.amap(isFactor, possible)                  
>>> obj
<processing.pool.MapResult object at 0x108efc450>
>>> print [x for x in obj.get() if x is not None]
[(1, 976), (2, 488), (4, 244), (8, 122), (16, 61)]
>>>
>>> # there's also parallel-python maps (blocking, iterative, and async) 
>>> from pathos.pp import ParallelPythonPool as PPool
>>> q = PPool(4)
>>> result = [x for x in q.map(isFactor, possible) if x is not None]
>>> print result
[(1, 976), (2, 488), (4, 244), (8, 122), (16, 61)]

另外,pathos 有一个具有相同接口的姊妹包,称为 pyina,它运行 mpi4py,但提供了在 MPI 中运行的并行映射,并且可以使用多个调度程序运行。

另一个优点是pathos 带有比标准python 中更好的序列化程序,因此它比multiprocessing 在序列化一系列函数和其他东西方面更有能力。您可以通过解释器完成所有操作。

>>> class Foo(object):
...   b = 1
...   def factory(self, a):
...     def _square(x):
...       return a*x**2 + self.b
...     return _square
... 
>>> f = Foo()
>>> f.b = 100
>>> g = f.factory(-1)
>>> p.map(g, range(10))
[100, 99, 96, 91, 84, 75, 64, 51, 36, 19]
>>> 

在此处获取代码:https://github.com/uqfoundation

【讨论】:

    【解决方案4】:

    这可以通过Ray 优雅地完成,该系统可让您轻松并行化和分发 Python 代码。

    要并行化您的示例,您需要使用@ray.remote 装饰器定义您的地图函数,然后使用.remote 调用它。这将确保远程函数的每个实例都将在不同的进程中执行。

    import ray
    
    ray.init()
    
    # Define the function to compute the factors of a number as a remote function.
    # This will make sure that a call to this function will run it in a different
    # process.
    @ray.remote
    def compute_factors(x):
        factors = [] 
        for i in range(1, x + 1):
           if x % i == 0:
               factors.append(i)
        return factors    
    
    # List of inputs.
    inputs = [67, 24, 18, 312]
    
    # Call a copy of compute_factors() on each element in inputs.
    # Each copy will be executed in a separate process.
    # Note that a remote function returns a future, i.e., an
    # identifier of the result, rather that the result itself.
    # This enables the calls to remote function to not be blocking,
    # which enables us to call many remote function in parallel. 
    result_ids = [compute_factors.remote(x) for x in inputs]
    
    # Now get the results
    results = ray.get(result_ids)
    
    # Print the results.
    for i in range(len(inputs)):
        print("The factors of", inputs[i], "are", results[i]) 
    

    multiprocessing 模块相比,使用Ray 有许多优点。特别是,相同的代码可以在单台机器上运行,也可以在一组机器上运行。有关 Ray 的更多优势,请参阅this related post

    【讨论】:

      猜你喜欢
      • 1970-01-01
      • 2016-03-04
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 2013-10-26
      • 2011-01-04
      • 1970-01-01
      • 1970-01-01
      相关资源
      最近更新 更多