如何使用Pandas和Dask优化两个超大CSV文件的合并操作

针对你遇到的Dask外连接大CSV耗时过长的问题，结合你的环境限制（不能用Parquet、必须CSV输入输出），我整理了几个亲测有效的优化方向，帮你提速：

1. 优化CSV读取的分区与类型推断 #

Dask读取CSV的默认设置往往不是最优的，调整以下参数能大幅减少读取时间，为后续合并打好基础：

• 手动指定分区大小：读取时用blocksize参数设置合适的分区（比如blocksize='128MB'），让每个分区的大小匹配你的内存和CPU核心数，避免过小的分区导致过多的任务调度开销，也避免过大的分区引发内存压力。
• 提前指定列类型：用dtype参数明确所有列的数据类型（比如dtype={'join_col1': 'int64', 'diff_col1': 'float64', ...}），彻底跳过Dask耗时的类型推断步骤。如果数据中缺少数值不多，还可以加上assume_missing=True，进一步减少类型检查的开销。
• 使用C引擎读取：确保engine='c'（默认值，但可以显式指定），C引擎的解析速度比Python引擎快数倍。

2. 优化连接执行策略 #

你试过基于索引合并但没效果？可能是索引设置或连接方法不对，试试这些调整：

• 正确设置复合索引：将用于外连接的4列全部设为两个DataFrame的索引（df = df.set_index(['col1', 'col2', 'col3', 'col4'])），然后合并时指定left_index=True, right_index=True。这样Dask能直接基于索引做连接，避免重复计算哈希值。
• 尝试排序合并方法：Dask默认用哈希连接，但对于大文件，排序合并（method='merge'）可能更高效，尤其是当你的连接键有一定有序性时。可以在merge时显式指定这个参数试试。
• 调整并行度与内存限制：初始化Dask Client时，设置合适的npartitions（建议为CPU核心数的2-4倍，比如24核设为48），让每个核心都能充分工作；同时设置memory_limit='45GB'，给系统留足内存空间，避免内存溢出触发磁盘交换（这会严重拖慢速度）。

3. 利用数据特征减少处理量 #

既然两个文件99.9%的列完全一致，只有2-3列有差异，完全可以减少不必要的数据处理：

• 只读取必要列：读取文件A时加载全部列，读取文件B时只加载连接键和那2-3个差异列，然后做外连接替换差异列。这样合并的数据量大幅减少，速度自然提升。比如：

  # 提前定义所有列的类型
  dtype_dict = {'join_col1': 'int64', 'join_col2': 'string', ..., 'diff_col1': 'float64'}
  
  # 读取文件A的全部列
  df_a = dd.read_csv('file_a.csv', dtype=dtype_dict, blocksize='128MB')
  # 读取文件B的连接键+差异列
  df_b = dd.read_csv('file_b.csv', usecols=['join_col1', 'join_col2', 'join_col3', 'join_col4', 'diff_col1', 'diff_col2'], dtype=dtype_dict, blocksize='128MB')
  # 执行外连接，区分两个文件的差异列
  merged = df_a.merge(df_b, on=['join_col1', 'join_col2', 'join_col3', 'join_col4'], how='outer', suffixes=('_a', '_b'))
  

• 避免重复列的冗余处理：你甚至可以以其中一个文件为基础，只合并差异列，而不是全列外连接，这能节省大量的内存和计算资源。

4. 优化CSV写入性能 #

合并后的写入环节也容易成为瓶颈，调整以下参数：

• 写入单个CSV文件：用to_csv时指定single_file=True，避免生成多个分区文件后再手动合并，减少额外的IO开销。
• 调整写入块大小：设置blocksize参数，让每个写入块更大，减少磁盘IO的次数。同时确保compression=None（如果不需要压缩），压缩会增加CPU开销。

5. 环境层面的小优化 #

• 确保使用SSD磁盘：CSV读写是IO密集型操作，SSD的随机读写速度远高于HDD，能大幅减少读写等待时间。
• 关闭无关进程：暂时关闭服务器上其他占用CPU、内存或磁盘资源的进程，让Dask独占硬件资源。

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