Dask Bag of dicts 到 Dask 数组

Question

我需要将 {'imgs': np.array(img_list), 'lables': np.array(label_list)} 的 dask.Bag 转换成两个单独的 dask.Array-s。为什么我创建 Bag 而不是直接去 Array？因为我正在通过 map() 多次处理那个 Bag；没能对 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 的情况下做到这一点？

想法：

我试过 Bag -> Delayed -> Array 但它没有用，因为数组形状有问题。
一个选项可能是将 Bag 作为文本文件转储到磁盘上，然后作为 dask 读取它。DataFrame/Array。示例：b_dict.map(json.dumps).to_textfiles("/path/to/data/*.json")
我可以不用一袋口述法，每袋有 2 袋 np.array，然后尝试 Bag -> Delayed -> Array。

还有其他想法吗？

Answer 1

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

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

那么这可以工作 (doc):

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

我推荐以下步骤：

制作两袋 numpy 数组（你必须使用 map 或 pluck 来提取你的 images 和 labels 值）
使用.map_partitions(np.stack)或.map_partitions(np.concatenate)（取决于你关心的形状）将你的每个分区变成一个单一的numpy数组
使用 .to_delayed
通过对每个延迟对象调用 dask.array.from_delayed 将每个延迟对象变成 dask.arrays
使用 da.stack 或 da.concatenate

Dask Bag of dicts 到 Dask 数组

Dask Bag of dicts to Dask array

python

etl

dask