将多处理和并行性应用于 numpy 函数

Question

我正在尝试并行化我的 numpy 代码以使其更快。原始的 numpy 数组 list_ 有超过 200 万个变量，因此很难计算这就是为什么我试图用 numpy 和多处理函数来计算我的计算。我如何利用下面的代码来处理多处理单元。我在下面有一个代码，我已经尝试过多处理，但我认为它不是很有效。函数 ym, xym ,std 是独立的，因此可以对这些函数进行多重处理，但是 mk,bk 依赖于 ym, xym ,std 函数。

原版 Python 版本

import numpy as np
number = 5
list_= np.array([457.334015,424.440002,394.795990,408.903992,398.821014,402.152008,435.790985,423.204987,411.574005,
404.424988,399.519989,377.181000,375.467010,386.944000,383.614990,375.071991,359.511993,328.865997,
320.510010,330.079010,336.187012,352.940002,365.026001,361.562012,362.299011,378.549011,390.414001,
400.869995,394.773010,382.556000])
ym = (np.convolve(list_, np.ones(shape=(number)), mode='valid')/number)[:-1]
xym = (np.convolve(list_, np.arange(number, 0, -1), mode='valid'))[:-1]
mk = (x_mean* ym- xym)/(np.power(x_mean,2)- x_squared_mean)
bk = ym - mk*x_mean
std = np.array([list_[i:i+number].std() for i in range(0, len(list_)-number)])

多处理版本

#x_mean and x_squared is used for the lin regressions and stand dev
x_mean = number/2*(1 + number)
x_squared_mean = number*(number+1)*(2*number+1)/6

def ym():   
    ym = (np.convolve(PC_list, np.ones(shape=(number)), mode='valid')/number)[:-1]
    print(ym)
def xym():
    xym = (np.convolve(PC_list, np.arange(number, 0, -1), mode='valid'))[:-1]
    print(xym)
def std():
    std = np.array([PC_list[i:i+number].std() for i in range(0, len(PC_list)-number)])
    print(std)
    
#setting up the multi processing vars     
p1 = multiprocessing.Process(target=ym)
p2 = multiprocessing.Process(target=xym)
p3 = multiprocessing.Process(target=std)

#running the multi processes 
p1.start()
p2.start()
p3.start()

p1.join()
p2.join()
p3.join()

xym()
ym()
std()

Answer 1

因为 mk 和 bk 确实依赖于 ym 和 xym，所以你无能为力，你必须先计算它们然后你可以在接下来的计算中使用。

我做了一个快速基准测试，看起来即使我有一个 2000 万的数组，卷积、mk 和 bk 都在 0.3 秒以下运行。代码中真正的瓶颈是 std 函数。那是因为您使用的 python for 循环很慢。

Numba 助你一臂之力

我建议调查 numba。它可以将您的常规 python 函数编译成优化的机器代码，并使它们运行非常快！它有一些限制，但对于您的情况，它会完美地工作。尝试将 std 函数更改为：

from numba import njit

@njit 
def std_numba(x, n):
    std = np.array([x[i:i+n].std() for i in range(0, len(x)-n)])
    return std

即使没有任何类型的多处理，所有计算运行在一个运行dom 数组上运行 200 万个浮点数。

将多处理和并行性应用于 numpy 函数

Applying multi processing and parallelism to numpy functions

python

numpy

process

multiprocessing

numpy-ndarray

Numba 助你一臂之力