ProcessPoolExecutor 不执行

Question

我尝试以比实际更快的速度获得 ARIMA 配置。

所以我使用 Iterate 方法来比较所有 ARIMA 组合以更好地选择。为此，我创建了一个迭代函数：

def difference(dataset, interval=1):
    diff = list()
    for i in range(interval, len(dataset)):
        value = dataset[i] - dataset[i - interval]
        diff.append(value)
    return np.array(diff)

# invert differenced value
def inverse_difference(history, yhat, interval=1):
    return yhat + history[-interval]

# evaluate an ARIMA model for a given order (p,d,q) and return RMSE
def evaluate_arima_model(dataset, arima_order):

    dataset = dataset.astype('float32')
    train_size = int(len(dataset) * 0.50)
    train, test = dataset[0:train_size], dataset[train_size:]
    
    history = [x for x in train]
    # make predictions
    predictions = list()
    for t in range(len(test)):
        # difference data
        months_in_year = maxlength
        diff = difference(history, months_in_year)
        model = ARIMA(diff, order=arima_order)
        model_fit = model.fit(trend='nc', disp=0)
        yhat = model_fit.forecast()[0]
        yhat = inverse_difference(history, yhat, months_in_year)
        predictions.append(yhat)
        history.append(test[t])
    # calculate out of sample error
    mse = mean_squared_error(test, predictions)
    rmse = sqrt(mse)
    return rmse

实际上，我用这种方法在几分钟内就能做到。但是现在不是我要使用逻辑的 API 的好时机。

# evaluate combinations of p, d and q values for an ARIMA model

def evaluate_models(dataset, p_values, d_values, q_values):
    dataset = dataset.astype('float32')
    train_size = int(len(dataset) * 0.50)
    train, test = dataset[0:train_size], dataset[train_size:]
    
    global best_score, best_cfg 
    best_score, best_cfg = float("inf"), None
    for p in p_values:
        for d in d_values:
            for q in q_values:
                order = (p,d,q)
                try:
                    mse = evaluate_arima_model(dataset, order)
                    if mse < best_score:
                        best_score, best_cfg = mse, order
                    print('ARIMA%s RMSE=%.3f' % (order,mse))
                except:
                    continue
        # print(best_cfg, best_score)
    print('Best ARIMA%s RMSE=%.3f' % (best_cfg, best_score))




# evaluate parameters
p_values = range(0, 7)
d_values = range(0, 3)
q_values = range(0, 7)
warnings.filterwarnings("ignore")
evaluate_models(data_train.values, p_values, d_values, q_values)

为了加速我想要使用 Multiprocessing 方法迭代 evaluate_arima_model 函数的过程。但是 ProcessPoolExecutor 不起作用，因为不打印任何结果

# evaluate combinations of p, d and q values for an ARIMA model
orders = []
def fill_orders( p_values, d_values, q_values):
    
    for p in p_values:
        for d in d_values:
            for q in q_values:
                order = (p,d,q)
                orders.append(order)

# fill orders array
p_values = range(0, 7)
d_values = range(0, 3)
q_values = range(0, 7)
warnings.filterwarnings("ignore")
fill_orders(p_values, d_values, q_values)

with concurrent.futures.ProcessPoolExecutor() as executor:
    results = [executor.submit(evaluate_arima_model, (dataset, order)) for order in orders]
    for f in concurrent.futures.as_completed(results):
        print(f.result())
        try:
            f.result()
        except:
            continue
        else:
            print(f.result())

Answer 1

我不希望您展示的第二个代码块能做任何事情。对于此代码：

# evaluate combinations of p, d and q values for an ARIMA model
orders = []
def evaluate_models( p_values, d_values, q_values):
    
    for p in p_values:
        for d in d_values:
            for q in q_values:
                order = (p,d,q)
                orders.append(order)

with concurrent.futures.ProcessPoolExecutor() as executor:
    results = [executor.submit(evaluate_arima_model, (dataset, order)) for order in orders]
    for f in concurrent.futures.as_completed(results):
        print(f.result())
        try:
            f.result()
        except:
            continue
        else:
            print(f.result())

orders 将始终为空，因为您这样声明它，然后从不调用 evaluate_models，或任何其他可能将对象放入 orders 的东西。因为orders 是空的，所以不会注册任何进程来运行，results 也将是空的，所以这段代码不会做任何事情。您的意思是在拨打with concurrent.futures.... 之前先拨打evaluate_models 吗？