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

Question

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

• 无法使用read_hdf() 从使用 h5py 创建的 dask hdf5 文件中读取。 我可以做些什么来导入数千个带有 dask 的 hdf5 文件并访问其中的多维数组？
• 我想创建一个 dask 数据帧，其中每个 2d 数组（从 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 格式。

当然，在某些领域，数据帧比数据集或数据数组更有意义，但是对于大于内存的数据，在它们之间进行转换可能会很棘手，因此如果您想要数据帧，第一种方法始终是一种选择。