如何创建迭代器逐行读取目录下多个Parquet文件？

已知可以用pyarrow读取单个Parquet文件的批量行，示例代码如下：

import pyarrow.parquet as pq

parquet_file = pq.ParquetFile('/mybigparquet.pq')
for i in parquet_file.iter_batches(batch_size=100):
    # 处理逻辑
    do something

现在需要处理同一目录/mybigparquetfiles/下的多个Spark生成的Parquet文件（命名格式如part_0001、part_0002等），要求：

• 创建一个迭代器，逐行扫描所有文件
• 每个文件仅被扫描一次
• 数据量过大无法存入内存，必须流式处理

请问是否可以实现该需求？

完全可以实现，pyarrow提供的ParquetDataset类专门用于处理目录下的多Parquet文件场景，结合流式批次迭代+行遍历，就能满足你的所有要求：全程仅加载当前批次数据到内存，每个文件仅被扫描一次，且能实现逐行迭代。

基础实现代码 #

import pyarrow.parquet as pq

def parquet_dir_row_iterator(dir_path, batch_size=100):
    # 初始化数据集，自动识别目录下所有Parquet文件
    dataset = pq.ParquetDataset(dir_path)
    # 流式遍历每个文件的批次
    for batch in dataset.to_batches(batch_size=batch_size):
        # 将批次转换为表格后，逐行生成数据
        table = batch.to_table()
        for row in table.to_pylist():
            yield row

# 使用示例
for row in parquet_dir_row_iterator('/mybigparquetfiles/'):
    # 处理单一行数据
    print(row)

关键细节说明 #

• ParquetDataset会自动识别目录下的所有Parquet文件，包括Spark生成的part_*格式文件，无需手动遍历文件名
• to_batches()方法会按指定batch_size流式读取数据，读完一个文件的所有批次后自动切换到下一个文件，天然保证每个文件仅被扫描一次
• 内存中始终只保留当前批次的数据，不会加载整个数据集，完全适配大内存占用场景
• 如果不需要转换为Python字典（追求更高性能），可以直接操作RecordBatch对象，比如通过batch.column('column_name')访问列数据后逐行处理

按文件名顺序读取的扩展实现 #

如果需要严格按照part_0001→part_0002的顺序读取文件，可以手动排序文件列表后传入数据集：

import os
import pyarrow.parquet as pq

def sorted_parquet_dir_row_iterator(dir_path, batch_size=100):
    # 获取目录下所有part文件并按名称排序
    file_paths = sorted([
        os.path.join(dir_path, f) 
        for f in os.listdir(dir_path) 
        if f.startswith('part_')
    ])
    # 基于排序后的文件列表初始化数据集
    dataset = pq.ParquetDataset(file_paths)
    for batch in dataset.to_batches(batch_size=batch_size):
        table = batch.to_table()
        for row in table.to_pylist():
            yield row

内容的提问来源于stack exchange，提问作者ℕʘʘḆḽḘ