Dask 函数 map_blocks 给出了 arange-compute_error

Question

我对 Dask 还很陌生，在使用函数 map_blocks 时遇到了一些问题。我正在尝试对二维数组的每个元素执行一个函数。我没有为索引 i 和 j 创建 2 个数组，而是创建了 1 个大小为 i * j 的数组。

ij = da.arange(n_users*n_ratings)
diff = da.map_blocks(compute_error, ij, dtype=np.float_).compute()

函数compute_error：

def compute_error(ij):
    i = int(ij/n_users)
    j = ij%n_users
    if not np.isnan(x[i,j]):
        return x[i, j] - np.dot(user_mat[j, :], ratings_mat[:, i])
    else:     
        return 0.0

矩阵x 看起来像：

1    Nan  Nan  Nan  5    2
Nan  Nan  Nan  Nan  4    Nan
Nan  3    Nan  Nan  4    Nan
Nan  3    Nan  Nan  Nan  Nan

矩阵user_mat (n_users X num_latent_features) 和ratings_mat(num_latent_features X num_ratings) ：

float float        float float float float
float float        float float float float
float float
float float 
float float 
float float 

我已阅读文档并搜索了 stackoverlow，但仍然无法解决以下问题：

KilledWorker                              Traceback (most recent call last)

<ipython-input-43-e670a6d660ce> in <module>
     12     # For each user-offer pair
     13     ij = da.arange(n_users*n_offers)
---> 14     diff = da.map_blocks(compute_error, ij, dtype=np.float_).compute()

c:\users\appdata\local\programs\python\python36\lib\site-packages\dask\base.py in compute(self, **kwargs)
    281         dask.base.compute
    282         
--> 283         (result,) = compute(self, traverse=False, **kwargs)
    284         return result
    285 

c:\users\appdata\local\programs\python\python36\lib\site-packages\dask\base.py in compute(*args, **kwargs)
    563         postcomputes.append(x.__dask_postcompute__())
    564 
--> 565     results = schedule(dsk, keys, **kwargs)
    566     return repack([f(r, *a) for r, (f, a) in zip(results, postcomputes)])
    567 

c:\users\appdata\local\programs\python\python36\lib\site-packages\distributed\client.py in get(self, dsk, keys, workers, allow_other_workers, resources, sync, asynchronous, direct, retries, priority, fifo_timeout, actors, **kwargs)
   2652                     should_rejoin = False
   2653             try:
-> 2654                 results = self.gather(packed, asynchronous=asynchronous, direct=direct)
   2655             finally:
   2656                 for f in futures.values():

c:\users\appdata\local\programs\python\python36\lib\site-packages\distributed\client.py in gather(self, futures, errors, direct, asynchronous)
   1967                 direct=direct,
   1968                 local_worker=local_worker,
-> 1969                 asynchronous=asynchronous,
   1970             )
   1971 

c:\users\appdata\local\programs\python\python36\lib\site-packages\distributed\client.py in sync(self, func, asynchronous, callback_timeout, *args, **kwargs)
    836         else:
    837             return sync(
--> 838                 self.loop, func, *args, callback_timeout=callback_timeout, **kwargs
    839             )
    840 

c:\users\appdata\local\programs\python\python36\lib\site-packages\distributed\utils.py in sync(loop, func, callback_timeout, *args, **kwargs)
    349     if error[0]:
    350         typ, exc, tb = error[0]
--> 351         raise exc.with_traceback(tb)
    352     else:
    353         return result[0]

c:\users\appdata\local\programs\python\python36\lib\site-packages\distributed\utils.py in f()
    332             if callback_timeout is not None:
    333                 future = asyncio.wait_for(future, callback_timeout)
--> 334             result[0] = yield future
    335         except Exception as exc:
    336             error[0] = sys.exc_info()

c:\users\appdata\local\programs\python\python36\lib\site-packages\tornado\gen.py in run(self)
    760 
    761                     try:
--> 762                         value = future.result()
    763                     except Exception:
    764                         exc_info = sys.exc_info()

c:\users\appdata\local\programs\python\python36\lib\site-packages\distributed\client.py in _gather(self, futures, errors, direct, local_worker)
   1826                             exc = CancelledError(key)
   1827                         else:
-> 1828                             raise exception.with_traceback(traceback)
   1829                         raise exc
   1830                     if errors == "skip":

KilledWorker: ("('arange-compute_error-71748aa3c524bc2a5b920efa05deec65', 2)", <Worker 'tcp://127.0.0.1:50070', name: 0, memory: 0, processing: 4>)

如果有任何更有效的方法来进行此计算，我也愿意接受建议。

Answer 1

而不是对 ij 数组进行操作，并将其值转换为索引 源数组，使用 dask 对实际的源数组进行操作。 它会大大加快。

我将源数组创建为：

1. 要创建 x 的源 (Numpy) 数组：

   arr = np.array([
       [1,      np.nan, np.nan, np.nan, 5,      2],
       [np.nan, np.nan, np.nan, np.nan, 4,      np.nan],
       [np.nan, 3,      np.nan, np.nan, 4,      np.nan],
       [np.nan, 3,      np.nan, np.nan, np.nan, np.nan]
   ])
   

2. x 数组（来自 arr）：

   x = da.from_array(arr, chunks=(2, 3))
   

   （我传递了 chunks 以避免将 x 创建为单块数组）。

3. user_mat 和 ratings_mat：

   user_mat = np.arange(1, 13, dtype='float').reshape(6, 2)
   ratings_mat = np.arange(2, 10, dtype='float').reshape(2, 4)
   

   我将它们创建为 Numpy 数组，但遵循 da 操作 将它们转换为 da 数组。

实际操作是：

result = da.where(da.notnull(x), da.subtract(x, da.dot(user_mat, ratings_mat).T), 0).compute()

步骤：

• da.notnull(x) - 结果选择标准（减法 或零），
• da.subtract(...) - 减法（第一个结果），
• 0 - 第二个结果（对于 x 中的 NaN 元素），
• da.where(...) - 计算内容的配方，
• compute() - 实际计算。

上述数据的结果是：

array([[ -13.,    0.,    0.,    0.,  -73.,  -92.],
       [   0.,    0.,    0.,    0.,  -93.,    0.],
       [   0.,  -41.,    0.,    0., -112.,    0.],
       [   0.,  -48.,    0.,    0.,    0.,    0.]])