python - 在 Dask 中，可以在 Dask 中将张量重塑为 2D 矩阵而不预先计算大小吗？

Question

在尝试创建一个能够在 Dask 上对标量函数进行矢量化的 python 基类时，我遇到了将张量重塑为 2D 矩阵的问题。解决这个问题将有助于创建可在 Numpy、Pandas 和 Dask 数据类型上互换操作的 sklearn 管道。

以下代码适用于 Dask 0.18.2但在 Dask 0.19.4和0.20.0上失败：

import dask
import dask.array
import dask.dataframe
import numpy
import pandas

def and1(x): return numpy.array([x, x+1], dtype=numpy.float32)

expected = numpy.array([[10, 11, 20, 21], 
                        [30, 31, 40, 41]], 
                       dtype=numpy.float32)

df = pandas.DataFrame.from_dict({
  'c1': [10, 30], 'c2': [20, 40]
})

ddf = dask.dataframe.from_pandas(df, npartitions=2)

# Dask generalized universal function that outputs 2 values per input value
guf = dask.array.gufunc(
    pyfunc=and1,
    signature='()->(n)',
    output_dtypes=numpy.float32,
    output_sizes={'n': 2},
    vectorize=True,
    allow_rechunk = False
)

da = guf(ddf)
da_reshaped = da.reshape((-1, numpy.prod(da.shape[1:])))
npa = da_reshaped.compute()

assert da.shape == (2, 2, 2)  # (input rows, input cols, outputs per cols)
numpy.testing.assert_array_equal(expected, npa)

在 Dask 0.19.4 和 0.20.0 中引发了一个 ValueError，因为s 形状reshape的第一个元素是（有关详细信息，请参阅堆栈跟踪）。daNaN

ValueErrorTraceback (most recent call last)
<ipython-input-847-ad2c41e1d88c> in <module>
     24 
     25 da = guf(ddf)
---> 26 da_r = da.reshape((-1, numpy.prod(da.shape[1:])))
     27 npa = da_r.compute()
     28 

/opt/conda/lib/python3.6/site-packages/dask/array/core.py in reshape(self, *shape)
   1398         if len(shape) == 1 and not isinstance(shape[0], Number):
   1399             shape = shape[0]
-> 1400         return reshape(self, shape)
   1401 
   1402     def topk(self, k, axis=-1, split_every=None):

/opt/conda/lib/python3.6/site-packages/dask/array/reshape.py in reshape(x, shape)
    160         if len(shape) == 1 and x.ndim == 1:
    161             return x
--> 162         missing_size = sanitize_index(x.size / reduce(mul, known_sizes, 1))
    163         shape = tuple(missing_size if s == -1 else s for s in shape)
    164 

/opt/conda/lib/python3.6/site-packages/dask/array/slicing.py in sanitize_index(ind)
     58                      _sanitize_index_element(ind.step))
     59     elif isinstance(ind, Number):
---> 60         return _sanitize_index_element(ind)
     61     elif is_dask_collection(ind):
     62         return ind

/opt/conda/lib/python3.6/site-packages/dask/array/slicing.py in _sanitize_index_element(ind)
     20     """Sanitize a one-element index."""
     21     if isinstance(ind, Number):
---> 22         ind2 = int(ind)
     23         if ind2 != ind:
     24             raise IndexError("Bad index.  Must be integer-like: %s" % ind)

ValueError: cannot convert float NaN to integer

是否有另一种方法可以在 Dask 0.20.0+ 中重塑 Dask 数组而不预先计算大小？ 如果是这样，重塑是否像在 Numpy 中那样是一个恒定时间操作？

我想创建一个矩阵 (shape = (R, C))，这样第一个轴不会更改，但所有后续轴都按"C"顺序合并（Dask 和 Numpy 中的默认值）。

（顺便说一句，我已经看到：Reshape a dask array (obtained from a dask dataframe column)）

标签： pythonpandasnumpydask