基于Python DataFrame通过简单查找高效编辑大输入文件

问题背景 #

• 处理6GB、5000万行的制表符分隔TXT文件，列包括id、amount、batch、transaction、sequence
• 已生成数百万行的runListDf，包含transaction和sequence-2列
• 当前通过pd.merge做左连接后过滤，merge阶段耗时极高；尝试set_index+join效率更差，需优化处理逻辑

核心优化思路 #

1. 从根源减少内存占用

5000万行的DataFrame内存开销是性能瓶颈的核心诱因，先从读取阶段优化：

• 指定精准数据类型：读取时给每个列分配最小可行的dtype，大幅降低内存占用：

  dtype_spec = {
      'id': 'category',          # 若id重复率高，用category比object省内存
      'amount': 'float32',       # 无需高精度时用float32替代float64
      'batch': 'category',       # 同理，批次号重复率高时用category
      'transaction': 'int64',    # 长数字用int64存储，避免转为object
      'sequence': 'int8'         # sequence是0/1这类小整数，用int8足够
  }
  inputFileDf = pd.read_csv(filename, header=None, index_col=False, sep='\t', 
                            names=['id','amount','batch','transaction','sequence'],
                            dtype=dtype_spec)
  

• 分块读取：若内存仍不足以加载全量数据，用chunksize参数分批次处理，避免一次性加载5000万行。

2. 替换Merge为集合查找（性能提升最显著）

你的需求本质是移除inputFileDf中(transaction, sequence)组合存在于runListDf的(transaction, sequence-2)组合中的行，完全不需要全量Merge：

• 先将runListDf的匹配组合转为哈希集合（查找时间复杂度O(1)）：

  # 用numpy数组转集合比pandas直接zip更快
  trans_arr = runListDf['transaction'].to_numpy()
  seq_arr = runListDf['sequence-2'].to_numpy()
  match_set = set(zip(trans_arr, seq_arr))
  

• 直接过滤inputFileDf：

  # 生成匹配键
  inputFileDf['match_key'] = list(zip(inputFileDf['transaction'], inputFileDf['sequence']))
  # 过滤掉存在于匹配集合中的行
  filtered_df = inputFileDf[~inputFileDf['match_key'].isin(match_set)]
  # 清理临时列
  filtered_df = filtered_df.drop('match_key', axis=1)
  # 写入结果
  filtered_df.to_csv(filename, header=None, index=None, sep='\t')
  

3. 分块处理的完整实现（内存不足时用）

如果无法加载全量inputFileDf，用分块读取逐批处理并追加输出：

import pandas as pd
import numpy as np

# 构建匹配集合
trans_arr = runListDf['transaction'].to_numpy()
seq_arr = runListDf['sequence-2'].to_numpy()
match_set = set(zip(trans_arr, seq_arr))

# 分块参数，可根据内存调整
chunk_size = 1_000_000
output_file = "filtered_result.txt"
first_write = True

# 逐块读取并处理
for chunk in pd.read_csv(filename, header=None, index_col=False, sep='\t',
                         names=['id','amount','batch','transaction','sequence'],
                         dtype={'id': 'category', 'amount': 'float32', 'batch': 'category',
                                'transaction': 'int64', 'sequence': 'int8'}):
    # 生成匹配键并过滤
    chunk['match_key'] = list(zip(chunk['transaction'], chunk['sequence']))
    filtered_chunk = chunk[~chunk['match_key'].isin(match_set)]
    filtered_chunk = filtered_chunk.drop('match_key', axis=1)
    
    # 写入文件，第一块覆盖，后续追加
    filtered_chunk.to_csv(output_file, header=None, index=None, sep='\t',
                          mode='w' if first_write else 'a')
    first_write = False

4. 其他细节优化

• 避免inplace=True：pandas的原地修改操作可能触发额外内存复制，优先用变量赋值替代。
• 用高效输出格式：若后续仍需用pandas处理，输出为Parquet格式（比TXT快数倍，占用空间仅为TXT的1/5左右），需安装pyarrow或fastparquet：

  filtered_df.to_parquet("filtered_result.parquet", index=False)
  

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