Dask的并行for循环比单核慢

我尝试过的

我有一个尴尬的并行for循环，其中我在两个嵌套的for循环中迭代90x360值并进行一些计算。我尝试dask.delayed按照本教程并行化for循环，尽管仅针对极少量的迭代进行了演示。

问题描述

我很惊讶地发现并行代码花费了2h 39分钟，而非并行定时花费了1h 54分钟，这意味着我做的事情根本上是错误的，或者任务图太大而无法处理？

设置信息

该测试是针对我的部分迭代进行的，即10 x 360，但是优化的代码应该能够处理90 x 360的嵌套迭代。我的迷你集群具有66个核心和256 GB的RAM，两个数据文件分别为4 GB和<1 GB。对于此任务，我multi-processing对使用vs的方法也感到困惑multi-threading。我认为在多个过程中运行类似于joblib默认实现的并行循环是每个循环在独立网格点上工作的方式。但是，这表明这样multi-threading做速度更快，如果没有GIL问题（我没有），则应首选此方法。因此，对于上述时间，我使用了dask.delay默认调度选项，该选项对单个进程使用多线程选项。

简化代码

import numpy as np
import pandas as pd
import xarray as xr
from datetime import datetime
from dask import compute, delayed

def add_data_from_small_file(lat):
    """ for each grid-point, get time steps from big-file as per mask, and
        compute data from small file for those time-steps
        Returns: array per latitude which is to be stacked
    """

    for lon in range(0,360):
        # get time steps from big file
        start_time = big_file.time.values[mask1[:, la, lo]] 
        end_time = big_file.time.values[[mask2[:,la,lo]]

        i=0  
        for t1, t2 in zip(start_time, end_time):
              # calculate value from small file for each time pair
              temp_var[i] = small_file.sel(t=slice(t1, t2)).median()
              i=i+1

         temp_per_lon[:, lon] = temp_var
     return temp_per_lon



if __name__ == '__main__':
    t1 = datetime.now()
    small_file = xr.open_dataarray('small_file.nc') # size < 1 GB, 10000x91
    big_file = xr.open_dataset('big_file.nc') # size = 4 GB, 10000x91x360

    delayed_values = [delayed(add_data_from_small_file)(lat) for lat in range(0,10)] # 10 loops for testing, to scale to 90 loops
    # have to delay stacking to avoid memory error
    stack_arr = delayed(np.stack)(delayed_values, axis=1) 
    stack_arr = stack_arr.compute()
    print('Total run time:{}'.format(datetime.now()-t1))