Python线性回归并行 - Scoop

Question

我试图使用Python的Scoop库并行运行从正态分布中随机生成的超过10000000个数据点（4个特征，1个目标变量）的线性回归。这是代码：

import pandas as pd
import numpy as np
import random
from scoop import futures
import statsmodels.api as sm
from time import time

def linreg(vals):
    global model
    model = sm.OLS(y_vals,X_vals).fit()
    return model
    print(model.summary())    

if __name__ == '__main__':

random.seed(42)
vals = pd.DataFrame(np.random.normal(loc = 3, scale = 100, size =(10000000,5)))
vals.columns = ['dep', 'ind1', 'ind2', 'ind3', 'ind4']
y_vals = vals['dep']
X_vals = vals[['ind1', 'ind2', 'ind3', 'ind4']]

bt = time()
model_vals = list(map(linreg, [1,2,3]))
mval = model_vals[0]
print(mval.summary())
serial_time = time() - bt

bt1 = time()
model_vals_1 = list(futures.map(linreg, [1,2,3]))
mval_1 = model_vals_1[0]
print(mval_1.summary())
parallel_time = time() - bt1

print(serial_time, parallel_time)`

但是，之后回归摘要确实是串行生成的 - 通过Python的标准映射函数 - 一个错误：

回溯（最近一次调用最后一次）：文件“C：\ Users \ niccolo.gentile \ AppData \ Local \ Continuum \ anaconda3 \ envs \ tensorenviron \ lib \ runpy.py”，第193行，在_run_module_as_main“main”，mod_spec）文件“C：\ Users \ niccolo.gentile \ AppData \ Local \ Continuum \ anaconda3 \ envs \ tensorenviron \ lib \ runpy.py”，第85行，在_run_code exec中（代码，run_globals）文件“C：\ Users \ niccolo.gentile \ AppData \ Local \ Continuum \ anaconda3 \ envs \ tensorenviron \ lib \ site-packages \ scoop \ bootstrap__main __。py“，第302行，在b.main（）文件”C：\ Users \ niccolo.gentile \ AppData \ Local \ Continuum \ anaconda3 \ envs \ tensorenviron \ lib \ site-packages \ scoop \ bootstrap__main __。py“，第92行，在主self.run（）文件”C：\ Users \ niccolo.gentile \ AppData \ Local \ Continuum \ anaconda3 \ envs \ tensorenviron \ lib \ site-packages \ scoop \ bootstrap__main __。py“，第290行，在运行的futures_startup（）文件”C：\ Users \ niccolo.gentile \ AppData \ Local \ Continuum \ anaconda3 \ envs \ tensorenviron \ lib \ site-packages \ scoop \ bootstrap__main __。py“，第271行，在futures_startup中运行run_name =”main“文件”C：\ Users \ n iccolo.gentile \ AppData \ Local \ Continuum \ anaconda3 \ envs \ tensorenviron \ lib \ site-packages \ scoop \ futures.py“，第64行，在_startup result = _controller.switch（rootFuture，* args，** kargs）文件“C：\ Users \ niccolo.gentile \ AppData \ Local \ Continuum \ anaconda3 \ envs \ tensorenviron \ lib \ site-packages \ scoop_control.py”，第253行，在runController中引发future.exceptionValue文件“C：\ Users \ niccolo .gentile \ AppData \ Local \ Continuum \ anaconda3 \ envs \ tensorenviron \ lib \ site-packages \ scoop_control.py“，第127行，在runFuture中。future.resultValue = future.callable（* future.args，** future.kargs）文件“C：\ Users \ niccolo.gentile \ AppData \ Local \ Continuum \ anaconda3 \ envs \ tensorenviron \ lib \ runpy.py”，第263行，在run_path中pkg_name = pkg_name，script_name = fname）文件“C：\ Users \ niccolo.gentile \ AppData \ Local \ Continuum \ anaconda3 \ envs \ tensorenviron \ lib \ runpy.py“，第96行，在_run_module_code mod_name，mod_spec，pkg_name，script_name）文件”C：\ Users \ niccolo.gentile \ AppData \ Local \ Continuum \ anaconda3 \ envs \ tensorenviron \ lib \ runpy.py“，li ne 85，在_run_code exec（代码，run_globals）文件“Scoop_map_linear_regression1.py”，第33行，在model_vals_1 = list（futures.map（linreg，[1,2,3]））文件“C：\ Users \ niccolo。 gentile \ AppData \ Local \ Continuum \ anaconda3 \ envs \ tensorenviron \ lib \ site-packages \ scoop \ futures.py“，第102行，在_mapGenerator中为将来的_waitAll（*期货）：文件”C：\ Users \ niccolo。 gentile \ AppData \ Local \ Continuum \ anaconda3 \ envs \ tensorenviron \ lib \ site-packages \ scoop \ futures.py“，第358行，在_waitAll中为f in _waitAny（future）：文件”C：\ Users \ niccolo.gentile \ _AppData \ Local \ Continuum \ anaconda3 \ envs \ tensorenviron \ lib \ site-packages \ scoop \ futures.py“，第335行，在_waitAny引发childFuture.exceptionValue NameError：name'y_vals'未定义

之后产生。这意味着代码在model_vals_1 = list(futures.map(linreg, [1,2,3]))处停止

请仔细注意，为了能够并行运行代码，必须从指定-m scoop参数的命令行启动它，如下所示：

python -m scoop Scoop_map_linear_regression1.py

实际上，如果它在没有-m scoop参数的情况下启动，它将不会被并行化并且确实会实际运行，而只是使用内置Python的map函数的两倍（因此，连续运行两次），就像你将得到的一样在警告中报告。也就是说，在启动它时没有指定-m scoop参数，future.map将被map替换，而目标是使用futures.map并行执行它。

这样澄清是为了避免人们通过简单地启动没有-m scoop参数的代码来回答他们解决了问题，如下所示：

Python Parallel Computing - Scoop

因此，这个问题被错误地搁置为主题，因为不再具有可再现性。

非常感谢提前，任何评论都非常感谢和欢迎。

Answer 1

解决方案是传递，作为futures.map的第二个参数（但不一定是map），只有[1]。

实际上，即使linreg函数不使用传递给map的第二个参数，它仍然确定linreg函数将运行多少次。例如，请考虑以下基本示例：

def welcome(x):
    print('Hello world!')

if __name__ == '__main__':
    a = list(map(welcome, [1,2]))

函数welcome实际上不需要任何参数，但输出仍然是

Hello world!
Hello world!

重复两次，即作为第二个参数传递的列表的长度。

在这种特定情况下，这意味着线性回归将按地图运行3次，尽管事实上回归输出只会出现一次，因为摘要是在地图外调用的。

关键是，不可能使用futures.map多次运行线性回归。问题是，显然，在第一次运行之后，它实际上删除了使用过的数据集，从而无法继续进行第二次和第三次运行，结果是

NameError：未定义名称“y_vals”

在Trace的末尾抛出。通过导航：scoop.futures source code可以看到这一点

没有仔细检查所有这些，但我想问题应该与greenlet切换器有关。