python

Question

我需要在数据分析 python 项目中同时使用 classes 和多处理功能，但我在 Google 上找不到很好的例子。

我的基本想法——这可能是错误的——是创建一个带有大变量的class（在我的例子中是一个pandas数据帧），然后定义一个方法计算一个操作（在本例中为总和）。

import multiprocessing
import time

class C:
    def __init__(self):
        self.__data = list(range(0, 10**7))

    def func(self, nums):
        return sum(nums)

    def start_multi(self):
        for n_procs in range(1, 4):
            print()
            time_start = time.clock()
            chunks = [self.__data[(i-1)*len(self.__data)// n_procs: (i)*len(self.__data)// n_procs] for i in range(1, n_procs+1)]
            pool = multiprocessing.Pool(processes=n_procs)
            results = pool.map_async(self.func, chunks )
            results.wait()
            pool.close()
            results = results.get()
            print(sum(results))
            print("n_procs", n_procs, "total time: ", time.clock() - time_start)

print('sum(list(range(0, 10**7)))', sum(list(range(0, 10**7))))
c = C()
c.start_multi()

代码无法正常工作：我得到以下打印输出

sum(list(range(0, 10**7))) 49999995000000

49999995000000
n_procs 1 total time:  0.45133500000000026

49999995000000
n_procs 2 total time:  0.8055279999999954

49999995000000
n_procs 3 total time:  1.1330870000000033

即计算时间不减反增。那么，这段代码中的错误是什么？

但我也担心 RAM 的使用，因为当创建变量块时，self.__data RAM 的使用会加倍。在处理多处理代码时，更具体地说，在这段代码中，是否有可能避免这种内存浪费？（我保证以后我会把所有东西都放在 Spark 上 :)）

Answer 1

看来这里有几件事在起作用：

分块操作非常慢。在我的计算机上，chunks 的生成占用了多进程情况下大约 16% 的时间。单进程、非池、版本没有这种开销。
您正在向您的流程发送大量数据。 chunks 数组是需要获取 pickled 并发送到新进程的范围的所有原始数据。与其发送所有原始数据，不如发送开始和结束索引会容易得多。
一般来说，如果您在 func 中放置计时器，您会发现大部分时间都没有花在那里。这就是为什么你没有看到加速。大部分时间花在分块、酸洗、分叉和其他开销上。

作为替代方案，您应该尝试将分块技术切换为仅计算开始和结束数字并避免发送太多数据。

接下来，我建议做一些比计算总和更难计算的事情。例如，您可以尝试计算素数。这是一个示例，我们使用 here 中的简单素数计算，并使用修改后的分块技术。否则，尽量保持代码不变。

import multiprocessing
import time
from math import sqrt; from itertools import count, islice

# credit to 
def isPrime(n):
    return n > 1 and all(n%i for i in islice(count(2), int(sqrt(n)-1)))

limit = 6
class C:
    def __init__(self):
        pass

    def func(self, start_end_tuple):
        start, end = start_end_tuple
        primes = []
        for x in range(start, end):
            if isPrime(x):
                primes.append(x)
        return len(primes)

    def get_chunks(self, total_size, n_procs):
        # start and end value tuples
        chunks = []

        # Example: (10, 5) -> (2, 0) so 2 numbers per process
        # (10, 3) -> (3, 1) or here the first process does 4 and the others do 3
        quotient, remainder = divmod(total_size, n_procs)
        current_start = 0
        for i in range(0, n_procs):
            my_amount = quotient
            if i == 0:
                # somebody needs to do extra
                my_amount += remainder
            chunks.append((current_start, current_start + my_amount))
            current_start += my_amount
        return chunks

    def start_multi(self):
        for n_procs in range(1, 4):
            time_start = time.clock()
            # chunk the start and end indices instead
            chunks = self.get_chunks(10**limit, n_procs)
            pool = multiprocessing.Pool(processes=n_procs)
            results = pool.map_async(self.func, chunks)
            results.wait()
            results = results.get()
            print(sum(results))
            time_delta = time.clock() - time_start
            print("n_procs {} time {}".format(n_procs, time_delta))

c = C()
time_start = time.clock()
print("serial func(...) = {}".format(c.func((1, 10**limit))))
print("total time {}".format(time.clock() - time_start))
c.start_multi()

这应该会导致多个进程的加速。假设您拥有它的核心。

python - class 字段和方法的多处理问题

python - multiprocessing issues with class fields and methods

multiprocessing

pool