带有分区的 dask read_parquet 目录的目录通配符答案

【问题标题】：directory globbing with partitioned dask read_parquet directories带有分区的 dask read_parquet 目录的目录通配符
【发布时间】：2021-08-13 00:35:44
【问题描述】：

我有一个分区气象站读数目录，我用pandas/pyarrow 编写。

c.to_parquet(path=f"data/{filename}.parquet", engine='pyarrow', compression='snappy', partition_cols=['STATION', 'ELEMENT'])

当我尝试使用如下所示的 glob 和谓词下推子句读取少量文件时

ddf= dd.read_parquet("data/*.parquet", engine='pyarrow', gather_statistics=True, filters=[('STATION', '==', 'CA008202251'), ('ELEMENT', '==', 'TAVG')], columns=['TIME','ELEMENT','VALUE', 'STATION'])

我得到一个索引错误

IndexError                                Traceback (most recent call last)
<timed exec> in <module>

/usr/local/lib/python3.9/site-packages/dask/dataframe/io/parquet/core.py in read_parquet(path, columns, filters, categories, index, storage_options, engine, gather_statistics, split_row_groups, read_from_paths, chunksize, aggregate_files, **kwargs)
    314         gather_statistics = True
    315 
--> 316     read_metadata_result = engine.read_metadata(
    317         fs,
    318         paths,

/usr/local/lib/python3.9/site-packages/dask/dataframe/io/parquet/arrow.py in read_metadata(cls, fs, paths, categories, index, gather_statistics, filters, split_row_groups, read_from_paths, chunksize, aggregate_files, **kwargs)
    540             split_row_groups,
    541             gather_statistics,
--> 542         ) = cls._gather_metadata(
    543             paths,
    544             fs,

/usr/local/lib/python3.9/site-packages/dask/dataframe/io/parquet/arrow.py in _gather_metadata(cls, paths, fs, split_row_groups, gather_statistics, filters, index, dataset_kwargs)
   1786 
   1787         # Step 1: Create a ParquetDataset object
-> 1788         dataset, base, fns = _get_dataset_object(paths, fs, filters, dataset_kwargs)
   1789         if fns == [None]:
   1790             # This is a single file. No danger in gathering statistics

/usr/local/lib/python3.9/site-packages/dask/dataframe/io/parquet/arrow.py in _get_dataset_object(paths, fs, filters, dataset_kwargs)
   1740         if proxy_metadata:
   1741             dataset.metadata = proxy_metadata
-> 1742     elif fs.isdir(paths[0]):
   1743         # This is a directory.  We can let pyarrow do its thing.
   1744         # Note: In the future, it may be best to avoid listing the

IndexError: list index out of range

我可以单独加载 parquet 目录

ddf= dd.read_parquet("data/2000.parquet", engine='pyarrow', gather_statistics=True, filters=[('STATION', '==', 'CA008202251'), ('ELEMENT', '==', 'TAVG')], columns=['TIME','ELEMENT','VALUE', 'STATION'])

dask/parquet/pyarrow 读取是否可以进行通配？

【问题讨论】：

标签： pandas dask parquet pyarrow

【解决方案1】：

在.to_parquet 中使用partition_cols 时，分区数据帧保存在单独的文件中，因此在您的情况下data/2000.parquet 可能是一个文件夹。

import pandas as pd
from os.path import isdir

# test dataframe
df = pd.DataFrame(range(3), columns=['a'])
df['b'] = df['a']
df['c'] = df['a']

# save without partitioning
df.to_parquet('test.parquet')
print(isdir('test.parquet')) # False

# save with partitioning
df.to_parquet('test_partitioned.parquet', partition_cols=['a', 'b'])
print(isdir('test_partitioned.parquet')) # True

作为解决此问题的一种方法，使用os.walk 或glob 构造parquet 文件的显式列表可能是一个很好的解决方案。请注意，如果有多个分区列，那么将有多个嵌套文件夹和 parquet 文件，因此简单的 glob 是不够的，您需要进行递归搜索。

或者，可以为每一年构造dask.dataframes，然后将它们与dd.concat 连接。

【讨论】：

【解决方案2】：

"data/*.parquet" 部分可能是导致您出现问题的原因。您需要提供不带* 的分区湖的根路径。

这是一个有效的示例代码 sn-p：

df = pd.DataFrame(
    [
        ["north america", "mexico", "carlos"],
        ["asia", "india", "ram"],
        ["asia", "china", "li"],
    ],
    columns=["continent", "country", "first_name"],
)
ddf = dd.from_pandas(df, npartitions=2)
ddf.to_parquet(
    "tmp/partition/2", engine="pyarrow", partition_on=["continent", "country"]
)

ddf = dd.read_parquet(
    "tmp/partition/2",
    engine="pyarrow",
    filters=[("continent", "==", "asia"), ("country", "==", "china")],
)

请注意，read_parquet 是在 "tmp/partition/2" 上调用的，而不是带有星号的目录。

【讨论】：