从Dask到Dask阵列的Dask袋

Question

我需要将dask.Bag of {'imgs': np.array(img_list), 'lables': np.array(label_list)}转换成两个独立的dask.Array-s。为什么我创建了袋子而不是直接去数组？因为我正在通过map()多次处理这个包，所以没有对Array做同样的处理。

现在，下面的代码适用于小型数据集，但显然不能用于更大的数据。

images_array = da.from_array(np.array([item['images'] for item in imgs_labels_bag]), chunks=chunksize)
labels_array = da.from_array(np.array([item['labels'] for item in imgs_labels_bag]), chunks=chunksize)

如何在不将objs转换为numpy的情况下做到这一点？

想法：

1. 我尝试过包->延迟->阵列，但它没有工作，因为数组形状的问题。
2. 一个选项可能是将包作为文本文件转储到磁盘上，然后将其作为dask.DataFrame/Array读取。示例：b_dict.map(json.dumps).to_textfiles("/path/to/data/*.json")
3. 与其有一袋dicts，我可以每个有2袋np.array，然后尝试each ->延迟->阵列。

还有其他想法吗？

Answer 1

如果item['images']是一维numpy数组，并且希望以以下方式平铺它们：

+---------------+
|item0['images']|
+---------------+
|item1['images']|
+---------------+
|item2['images']|
+---------------+    

然后这个可以工作(文档)：

import dask.bag as db
import numpy as np
import dask.array as da
b = db.from_sequence([{'img':np.arange(10)}]*4)
s = da.stack([item['img'] for item in b], axis=0)
print(s.compute())

结果：

[[0 1 2 3 4 5 6 7 8 9]
 [0 1 2 3 4 5 6 7 8 9]
 [0 1 2 3 4 5 6 7 8 9]
 [0 1 2 3 4 5 6 7 8 9]]

Answer 2

我建议采取以下步骤：

1. 制作两袋numpy数组(您必须使用映射或拔出来提取images和labels值)
2. 使用.map_partitions(np.stack)或.map_partitions(np.concatenate) (取决于您所关心的形状)将每个分区转换为单个numpy数组
3. 使用dask.delayed将分区转换为.to_delayed对象
4. 通过对每个延迟对象调用dask.arrays，将每个延迟对象转换为dask.array.from_delayed
5. 使用dask.array或da.concatenate将这些dask数组叠加或连接到一个单一的da.concatenate中。