Pandas基于条件删除大数据集行的速度优化方案咨询

针对你1200个160万行数据集、1400余种筛选条件的场景，结合你提到的df.drop效率不如当前布尔索引的情况，以下是具体的提速优化方案：

• 强制用向量化操作替代逐行循环
  别用df.apply这类逐行处理的方法，直接用pandas/NumPy的向量化计算——这是提升速度的核心。以你给出的示例条件为例：

  # 向量化计算每行处于30-50区间的列数
  valid_col_count = ((df >= 30) & (df <= 50)).sum(axis=1)
  # 直接生成布尔掩码筛选
  filtered_df = df[(valid_col_count >= 2) & (valid_col_count <= 6)]
  

  向量化操作是底层C级循环，比Python级逐行循环快几个数量级。

• 压缩数据类型降低内存开销
  内存占用过高会严重拖慢计算和IO速度，优先优化数据类型：

  • 数值型列：在精度允许的情况下，将float64转float32、int64转int32：

    num_cols = df.select_dtypes(include=['int64', 'float64']).columns
    df[num_cols] = df[num_cols].apply(pd.to_numeric, downcast='integer' if 'int' in str(df[num_cols].dtype) else 'float')
    

  • 分类列：重复值多的列转category类型，能大幅减少内存：

    cat_cols = df.select_dtypes(include=['object']).columns
    df[cat_cols] = df[cat_cols].astype('category')
    

• 分块处理单个大文件
  单个160万行的文件如果一次性加载内存压力大，用chunksize分块读取处理，避免内存溢出：

  chunk_size = 100000  # 可根据内存调整块大小
  output_file = 'filtered_result.csv'
  # 初始化写入表头
  first_write = True
  for chunk in pd.read_csv('large_data.csv', chunksize=chunk_size):
      valid_col_count = ((chunk >= 30) & (chunk <= 50)).sum(axis=1)
      filtered_chunk = chunk[(valid_col_count >= 2) & (valid_col_count <= 6)]
      filtered_chunk.to_csv(output_file, index=False, mode='a', header=first_write)
      first_write = False
  

• 并行处理1200个独立数据集
  因为1200个文件是独立的，用多进程并行处理可以充分利用CPU核心：

  from concurrent.futures import ProcessPoolExecutor
  import pandas as pd
  import os
  
  def process_file(file_path):
      # 读取文件
      df = pd.read_csv(file_path)
      # 执行筛选逻辑
      valid_col_count = ((df >= 30) & (df <= 50)).sum(axis=1)
      filtered_df = df[(valid_col_count >= 2) & (valid_col_count <= 6)]
      # 保存结果
      output_path = os.path.join('filtered', os.path.basename(file_path))
      filtered_df.to_csv(output_path, index=False)
      return f"Processed {file_path}"
  
  # 生成文件列表
  file_list = [os.path.join('data_dir', f) for f in os.listdir('data_dir') if f.endswith('.csv')]
  # 并行处理，max_workers设为CPU核心数的70%-80%
  with ProcessPoolExecutor(max_workers=6) as executor:
      results = executor.map(process_file, file_list)
  

  注意：如果是IO密集型（读写文件多），可以用ThreadPoolExecutor替代，开销更小。

• 用NumPy底层操作进一步压榨性能
  要是pandas的向量化还不够快，直接用NumPy数组操作，减少pandas的封装开销：

  import numpy as np
  
  # 转成NumPy数组
  data_np = df.to_numpy()
  # 计算每行符合条件的元素数
  count_np = np.sum((data_np >= 30) & (data_np <= 50), axis=1)
  # 生成掩码并筛选
  mask_np = (count_np >= 2) & (count_np <= 6)
  filtered_df = pd.DataFrame(data_np[mask_np], columns=df.columns)
  

• 预封装重复条件逻辑
  针对1400余种条件，把重复的筛选逻辑封装成函数，避免重复代码和计算开销：

  def filter_by_range_count(df, lower=30, upper=50, min_cnt=2, max_cnt=6):
      cnt = ((df >= lower) & (df <= upper)).sum(axis=1)
      return df[(cnt >= min_cnt) & (cnt <= max_cnt)]
  
  # 调用时直接传参数
  filtered_df = filter_by_range_count(df, lower=20, upper=40, min_cnt=3, max_cnt=5)
  

• 只加载必要的列
  如果很多列不需要参与筛选也不需要保留，读取时直接指定usecols，减少数据量：

  # 只加载用于筛选的列和需要保留的列
  cols_needed = ['keep_col1', 'keep_col2', 'filter_col1', 'filter_col2']
  df = pd.read_csv('large_data.csv', usecols=cols_needed)
  

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