将 HDF 文件加载到 Python Dask DataFrames 列表中

Posted 2023-03-11

【中文标题】将 HDF 文件加载到 Python Dask DataFrames 列表中

【英文标题】：Load HDF file into list of Python Dask DataFrames

【发布时间】：2017-10-05 06:33:08

【问题描述】：

我有一个 HDF5 文件，我想将它加载到 Dask DataFrames 列表中。我已经使用 Dask pipeline approach 的缩写版本之后的循环来设置它。代码如下：

import pandas as pd
from dask import compute, delayed
import dask.dataframe as dd
import os, h5py

@delayed
def load(d,k):
    ddf = dd.read_hdf(os.path.join(d,'Cleaned.h5'), key=k)
    return ddf

if __name__ == '__main__':      
    d = 'C:\Users\User\FileD'
    loaded = [load(d,'/DF'+str(i)) for i in range(1,10)]

    ddf_list = compute(*loaded)
    print(ddf_list[0].head(),ddf_list[0].compute().shape)

我收到此错误消息：

C:\Python27\lib\site-packages\tables\group.py:1187: UserWarning: problems loading leaf ``/DF1/table``::

  HDF5 error back trace

  File "..\..\hdf5-1.8.18\src\H5Dio.c", line 173, in H5Dread
    can't read data
  File "..\..\hdf5-1.8.18\src\H5Dio.c", line 543, in H5D__read
    can't initialize I/O info
  File "..\..\hdf5-1.8.18\src\H5Dchunk.c", line 841, in H5D__chunk_io_init
    unable to create file chunk selections
  File "..\..\hdf5-1.8.18\src\H5Dchunk.c", line 1330, in H5D__create_chunk_file_map_hyper
    can't insert chunk into skip list
  File "..\..\hdf5-1.8.18\src\H5SL.c", line 1066, in H5SL_insert
    can't create new skip list node
  File "..\..\hdf5-1.8.18\src\H5SL.c", line 735, in H5SL_insert_common
    can't insert duplicate key

End of HDF5 error back trace

Problems reading the array data.

The leaf will become an ``UnImplemented`` node.
  % (self._g_join(childname), exc))

消息中提到了重复的键。我迭代了前 9 个文件以测试代码，并且在循环中，我使用每个迭代来组装一个不同的密钥，用于dd.read_hdf。在所有迭代中，我保持文件名相同 - 只有密钥被更改。

我需要使用dd.concat(list,axis=0,...) 来垂直连接文件的内容。我的方法是先将它们加载到一个列表中，然后将它们连接起来。

我已经安装了PyTables 和h5Py 并且有Dask 版本0.14.3+2。

有了 Pandas 0.20.1，我似乎可以做到这一点：

for i in range(1,10):
    hdf = pd.HDFStore(os.path.join(d,'Cleaned.h5'),mode='r')
    df = hdf.get('/DF' .format(i))
    print df.shape
    hdf.close()

有没有办法可以将此 HDF5 文件加载到 Dask DataFrames 列表中？还是有另一种将它们垂直连接在一起的方法？

【参考方案1】：

Dask.dataframe 已经是懒惰的了，所以没有必要使用 dask.delayed 让它更懒惰。你可以反复拨打dd.read_hdf：

ddfs = [dd.read_hdf(os.path.join(d,'Cleaned.h5'), key=k)
        for k in keys]

ddf = dd.concat(ddfs)

【讨论】：

我错过了。谢谢！

是否可以在同一管道中混合使用delayed 和non-delayed 函数？

请参阅these docs，了解如何在延迟值和 dask.dataframes 之间进行转换。没有理由在惰性函数中嵌套惰性函数。

您还可以将路径列表直接传递给dd.read_hdf的一次调用。