如何高效将S3存储的parquet文件按块迭代处理

你当前的实现最大的问题是需要先将全量Parquet数据加载到内存再切片，数据量较大时会有OOM风险，也浪费IO和内存资源，以下是时空效率更优的实现方案：

方案1：直接使用fastparquet原生迭代能力（无需更换依赖） #

fastparquet本身支持按需读取部分行，不需要先转全量DataFrame再拆分，同时可以删除原有冗余的S3FileSystem初始化逻辑：

import s3fs
import fastparquet as fp

s3 = s3fs.S3FileSystem()

bucket, path = 'mybucket', 'mypath'
root_dir_path = f'{bucket}/{path}'
s3_path = f"{root_dir_path}/*.parquet"
all_paths_from_s3 = s3.glob(path=s3_path)

fp_obj = fp.ParquetFile(all_paths_from_s3, open_with=s3.open, root=root_dir_path)
chunksize = 1000

# 按需读取指定范围的行，不会加载全量数据到内存
for i in range(0, fp_obj.count(), chunksize):
    chunk = fp_obj.to_pandas(index=slice(i, i+chunksize))
    # 你的分块处理逻辑
    print(len(chunk))

如果你的业务逻辑可以接受按Parquet原生行组大小处理，直接迭代行组性能更高：

for rg in fp_obj.iter_row_groups():
    chunk = rg.to_pandas()
    # 你的分块处理逻辑
    print(len(chunk))

方案2：使用PyArrow Dataset接口（性能更优，推荐） #

PyArrow的数据集接口对云存储读取、下推优化的支持更好，是目前处理大规模Parquet数据的主流选择：

import s3fs
import pyarrow.dataset as ds

s3 = s3fs.S3FileSystem()
bucket, path = 'mybucket', 'mypath'
s3_path = f"s3://{bucket}/{path}"

# 直接扫描S3路径下的Parquet数据集，自动识别分区结构
dataset = ds.dataset(s3_path, filesystem=s3, format="parquet")

# 按指定大小迭代，支持列投影、谓词下推，大幅降低IO开销
for batch in dataset.to_batches(
    # 只读取需要的列，不要全列加载
    columns=["col1", "col2", "col3"],
    # 可选：过滤符合条件的行，直接跳过不需要的数据
    filter=ds.field("col3") > 10,
    batch_size=1000
):
    chunk = batch.to_pandas()
    # 你的分块处理逻辑
    print(len(chunk))

优化收益说明 #

• 空间效率：峰值内存占用从全量数据大小降到单块数据大小，避免OOM风险，适合处理GB/TB级别的数据集
• 时间效率：跳过全量数据加载、切片的额外开销，结合列投影、谓词下推可以减少90%以上的无效IO，整体处理速度提升非常明显

内容的提问来源于stack exchange，提问作者ProgramSpree