使用Dask DataFrame保存Parquet时生成文件夹而非文件的解决方案

问题根源 #

Dask默认将DataFrame保存为Parquet数据集（文件夹形式，包含多分区文件与元数据），这是为适配分布式多分区场景。要生成单个Parquet文件，需针对单个分区的Dask DataFrame做参数配置，避免转Pandas带来的性能损耗。

高效解决方案 #

方案1：手动分块+单文件保存

确保处理后的chunk为单个分区，配合single_file=True参数直接生成单个Parquet文件：

def splitting_files(df, chunk_size):
    total_rows = len(df)
    row_indices = list(range(0, total_rows, chunk_size))
    if row_indices[-1] < total_rows:
        row_indices.append(total_rows)

    for i in range(len(row_indices) - 1):
        start_idx = row_indices[i]
        end_idx = row_indices[i+1]
        
        chunk_df = df.loc[start_idx: end_idx]
        chunk_df = processing_files(chunk_df)
        
        # 强制将chunk转为单个分区（避免多分区导致生成文件夹）
        chunk_df = chunk_df.repartition(npartitions=1)
        # 每个chunk指定独立文件名，而非文件夹路径
        output_path = f"some intermediate folder/chunk_{i}.parquet"
        # 启用single_file参数生成单个Parquet文件
        chunk_df.to_parquet(
            output_path,
            engine='fastparquet',
            index=False,
            single_file=True
        )

方案2：利用Dask原生分区处理（更高效）

跳过手动按行切片，直接遍历Dask DataFrame的原生分区，避免len(df)触发全量计算的开销：

def process_by_partition(df):
    for idx, partition in enumerate(df.partitions):
        # 处理单个分区
        chunk_df = processing_files(partition)
        # 确保为单个分区（部分操作可能导致分区数变化）
        chunk_df = chunk_df.repartition(npartitions=1)
        # 保存为单个Parquet文件
        output_path = f"some intermediate folder/chunk_{idx}.parquet"
        chunk_df.to_parquet(
            output_path,
            engine='fastparquet',
            index=False,
            single_file=True
        )

关键注意事项 #

• single_file=True要求Dask版本≥2021.06.0，且依赖fastparquet或pyarrow引擎
• 单个分区的chunk大小需控制在内存可承受范围内（避免内存溢出）
• 后续合并时，直接通过dd.read_parquet("some intermediate folder/*.parquet")读取所有单个Parquet文件即可

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