如何使用 dask 并行导入 hdf5 数据并创建 dask 数据框？

Question

我完全被卡住了，因此我正在寻找好的建议。我的目标是并行读出许多 hdf5 文件，提取其中的 multi-dim 数组，并将每个数组存储在 dask 数据帧的一行中，恰好是一个单元格。我不选择 pandas df，因为我认为它会太大。

无法使用 read_hdf() 读取使用 h5py 创建的 dask hdf5 文件。我可以做些什么来并行导入数千个带有 dask 的 hdf5 文件并访问其中的多维数组？
我想创建一个 dask 数据帧，其中每个二维数组（从 hdfs 内的 n-dim 数组中提取）存储在 dask 数据帧的一个单元格中。因此，行数对应于所有文件中找到的总数组数，此处为 9。我将数组存储在一列中。
将来我想将更多列和其他数据附加到这个 dask 数据框。我想用另一个 Python 库对数组进行操作，并将结果存储在 dask 数据框的其他列中。数据框应包含我提取和操作的所有信息。我还想从其他 hdf5 文件中添加数据。就像一个小型数据库。这合理吗？
我可以并行工作，因为数组彼此独立。

请问您是如何实现这一点的？ xarray 也被建议给我，但我不知道什么是最好的方法。早些时候我试图收集多维 dask 数组中的所有数组，但是只有 ndim=2.

才能转换为数据框

感谢您的建议。祝你有个美好的一天。

import numpy as np
import h5py
import dask.dataframe as dd
import dask.array as da
import dask
print('This is dask version', dask.__version__)


ra=np.ones([10,3199,4000])

print(ra.shape)
file_list=[]
for i in range(0,4):
    #print(i)
    fstr='data_{0}.h5'.format(str(i))
    #print(fstr)
    hf = h5py.File('./'+fstr, 'w')
    hf.create_dataset('dataset_{0}'.format(str(i)), data=ra)
    hf.close()
    file_list.append(fstr)

!ls

print(file_list)

for i,fn in enumerate(file_list):
    dd.read_hdf(fn,key='dataset_{0}'.format(str(i))) #breaks here

Answer 1

您可以使用 dask.distributed 将数据预处理为数据帧，然后使用 dask.dataframe.from_delayed.

将期货转换为单个 dask.dataframe

from dask.distributed import Client
import dask.dataframe as dd

client = Client()

def preprocess_hdf_file_to_dataframe(filepath):
    # process your data into a dataframe however you want, e.g.
    with xr.open_dataset(filepath) as ds:
        return ds.to_dataframe()

files = ['file1.hdf5', 'file2.hdf5']

futures = client.map(preprocess_hdf_file_to_dataframe, files)
df = dd.from_delayed(futures)

也就是说，这似乎是 xarray 的完美用例，它可以读取 HDF5 文件并在本地使用 dask，例如

ds = xr.open_mfdataset(files)

此数据集类似于 dask.dataframe，因为它包含对从文件中读取的 dask.arrays 的引用。但是 xarray 是为原生处理 N 维数组而构建的，并且可以更自然地与 HDF5 格式一起工作。

虽然在某些方面数据帧比 Dataset 或 DataArray 更有意义，但对于大于内存的数据，它们之间的转换可能会很棘手，因此如果您想要数据帧，第一种方法始终是一个选择.

如何使用 dask 并行导入 hdf5 数据并创建 dask 数据框？

How to import hdf5 data with dask in parallel and create a dask dataframe?

python

parallel-processing

dask

dask-dataframe