使用Dask合并大量CSV文件遇性能问题，Pandas却正常运行求排查

你的Dask代码效率低下甚至内存溢出的核心问题在于大量极小文件导致的分区爆炸，以及不必要的全量内存加载：

1. 分区数量过大，调度开销远超计算成本
   dd.read_csv(files)会为每个CSV文件创建一个独立分区，7万个小文件就对应7万个分区。Dask的调度器需要为每个分区分配任务、跟踪状态、传输数据，这种调度开销对于执行时间极短的小任务来说，会完全盖过并行计算带来的收益，导致整体速度远慢于串行。

2. 全量内存加载触发内存瓶颈
   df.compute()会将整个Dask DataFrame的所有数据加载到单进程内存中，本质和串行版本的pd.concat类似，但Dask额外的分区元数据、中间缓存会进一步增加内存占用，最终触发swap和GC警告。

针对大量小gzip压缩CSV的场景，核心思路是减少分区数量，并避免全量内存加载，以下是具体实现：

方案1：批量分组处理小文件，控制分区规模 #

将小文件分组，每组合并为一个较大的DataFrame分区，再用Dask并行处理这些分组：

import dask.dataframe as dd
from dask import delayed
import pandas as pd

# 将文件按每组N个分组，建议每组解压后大小在50-100MB左右（这里按5000个一组，解压后约45MB）
group_size = 5000
file_groups = [files[i:i+group_size] for i in range(0, len(files), group_size)]

def process_file_group(file_group):
    """处理单个文件组，合并为一个DataFrame"""
    dfs = []
    for f in file_group:
        df = pd.read_csv(f)
        # 执行你的索引列添加逻辑
        df["universal_index"] = ...  # 替换为你的实际代码
        dfs.append(df)
    return pd.concat(dfs, ignore_index=True)

# 生成延迟执行的分组任务
delayed_dfs = [delayed(process_file_group)(group) for group in file_groups]

# 转换为Dask DataFrame
ddf = dd.from_delayed(delayed_dfs)

# 直接写入Parquet，无需全量加载到内存
ddf.to_parquet(
    "final_merged.parquet",
    compression="gzip",  # 或使用snappy等更高效的压缩格式
    write_index=False
)

方案2：先合并小文件为大文件（可选） #

如果后续还有类似任务，可以先将小gzip文件合并为较大的压缩文件（比如每200MB一个），再用Dask读取：

# 示例：将当前目录下的小csv.gz文件合并为多个大文件（每1000个一组）
count=0
batch=0
for f in *.csv.gz; do
    cat "$f" >> "batch_$batch.csv.gz"
    ((count++))
    if [ $count -eq 1000 ]; then
        ((batch++))
        count=0
    fi
done

之后用dd.read_csv("batch_*.csv.gz")读取，此时分区数大幅减少，Dask的并行效率会显著提升。

• 控制分区大小：Dask推荐每个分区的解压后大小在100MB-1GB之间，过小的分区会导致调度浪费，过大则会降低并行度。
• 避免全量compute()：直接使用Dask的to_parquet可以分块写入磁盘，无需将所有数据加载到内存，从根源避免内存溢出。
• gzip的局限性：gzip是不可分割的压缩格式，Dask无法并行读取单个gzip文件，因此合并小文件是提升效率的必要步骤。如果后续可以选择zstd、brotli等可分割压缩格式，会进一步优化并行读取性能。

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