如何优化Pandas/PySpark中同后缀不同前缀列的数值对比效率？

现有一个约4500列、50000行的Pandas DataFrame，列分为以'A'和'B'开头的两类，示例列如下：

[id, A.sellervalue, A.buyervalue, A.ratingvalue, B.sellervalue, B.buyervalue, B.ratingvalue]

我写了一段代码用来对比同后缀不同前缀的列（比如A.sellervalue和B.sellervalue），生成记录不匹配行、列名及对应值的新DataFrame，但这段代码效率太低，求基于Pandas方法或PySpark的优化方案。

原代码如下：

def find_mismatch_common_columns_values(df, A_cols_without_prefix, B_cols_without_prefix):
    set_A_cols_without_prefix = set(A_cols_without_prefix)
    set_B_cols_without_prefix = set(B_cols_without_prefix)
    common_columns = list(set_A_cols_without_prefix.intersection(set_B_cols_without_prefix))
    
    print("\n Size of common columns between both ",len(common_columns))
        
    result_list = []    
    for col in common_columns:
        A_col = f'A.{col}'
        B_col = f'B.{col}'

        for idx, row in df.iterrows():
            
            A_value = row[A_col]
            B_value = row[B_col]
            
            if isinstance(A_value, float) :
                A_value = round(float(A_value), 6)
            
            if isinstance(B_value, float):
                B_value = round(float(B_value), 6)
                
            
            if A_value != B_value and (not pd.isnull(A_value) and not pd.isnull(B_value)):
                result_list.append({
                    'row': idx,
                    'column_name': col,
                    'A_value': A_value,
                    'B_value': B_value
                })

    
    result_df = pd.DataFrame(result_list)
    return result_df

Pandas优化方案 #

原代码最大的问题是嵌套循环+iterrows()逐行遍历，完全浪费了Pandas的向量化优势，换成以下方案能大幅提速：

1. 向量化批量处理版本 #

import pandas as pd

def find_mismatch_pandas_optimized(df, A_cols_without_prefix, B_cols_without_prefix):
    common_cols = list(set(A_cols_without_prefix) & set(B_cols_without_prefix))
    print(f"\n共同列数量: {len(common_cols)}")
    
    result_dfs = []
    for col in common_cols:
        a_col = f'A.{col}'
        b_col = f'B.{col}'
        
        # 复制列数据，避免修改原表
        a_vals = df[a_col].copy()
        b_vals = df[b_col].copy()
        
        # 对浮点类型列统一保留6位小数（整列操作，比逐元素判断快N倍）
        if pd.api.types.is_float_dtype(a_vals):
            a_vals = a_vals.round(6)
        if pd.api.types.is_float_dtype(b_vals):
            b_vals = b_vals.round(6)
        
        # 用布尔掩码筛选不匹配且非空的行
        mask = (a_vals != b_vals) & (~a_vals.isnull()) & (~b_vals.isnull())
        mismatched_idx = df[mask].index
        
        if not mismatched_idx.empty:
            temp_df = pd.DataFrame({
                'row': mismatched_idx,
                'column_name': col,
                'A_value': a_vals[mismatched_idx],
                'B_value': b_vals[mismatched_idx]
            })
            result_dfs.append(temp_df)
    
    # 合并所有结果
    return pd.concat(result_dfs, ignore_index=True) if result_dfs else pd.DataFrame()

• 核心优化：用整列的round()和布尔掩码替代逐行判断，利用Pandas底层的C语言实现，彻底避免Python级别的循环开销。

2. 自动提取列后缀（可选） #

如果列名严格遵循A.xxx和B.xxx格式，不用手动传入后缀列表，自动提取更省心：

# 自动从列名中提取A、B的后缀
a_suffixes = [col.split('.')[1] for col in df.columns if col.startswith('A.')]
b_suffixes = [col.split('.')[1] for col in df.columns if col.startswith('B.')]
common_cols = list(set(a_suffixes) & set(b_suffixes))

PySpark优化方案 #

如果后续数据量还会增长，单节点Pandas扛不住的话，用PySpark的分布式计算更合适：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, round, lit, when

def find_mismatch_pyspark(spark, df):
    # 自动提取共同列后缀
    a_suffixes = [col.split('.')[1] for col in df.columns if col.startswith('A.')]
    b_suffixes = [col.split('.')[1] for col in df.columns if col.startswith('B.')]
    common_cols = list(set(a_suffixes) & set(b_suffixes))
    print(f"\n共同列数量: {len(common_cols)}")
    
    result_dfs = []
    for col_suffix in common_cols:
        a_col = f'A.{col_suffix}'
        b_col = f'B.{col_suffix}'
        
        # 处理浮点类型的四舍五入，非浮点列直接保留原值
        a_val = when(col(a_col).cast("float").isNotNull(), round(col(a_col), 6)).otherwise(col(a_col))
        b_val = when(col(b_col).cast("float").isNotNull(), round(col(b_col), 6)).otherwise(col(b_col))
        
        # 筛选不匹配且非空的行，提取需要的字段
        mismatched_df = df.filter(
            (a_val != b_val) & (col(a_col).isNotNull()) & (col(b_col).isNotNull())
        ).select(
            col("id").alias("row"),  # 用id作为行标识，如果没有id可以用monotonically_increasing_id()生成索引
            lit(col_suffix).alias("column_name"),
            a_val.alias("A_value"),
            b_val.alias("B_value")
        )
        
        result_dfs.append(mismatched_df)
    
    # 合并所有结果DataFrame
    if not result_dfs:
        return spark.createDataFrame([], "row int, column_name string, A_value string, B_value string")
    return result_dfs[0].unionAll(*result_dfs[1:])

# 使用示例
spark = SparkSession.builder.appName("ColumnMismatchCheck").getOrCreate()
# 假设df是你的PySpark DataFrame
result_spark_df = find_mismatch_pyspark(spark, df)
result_spark_df.show()

• 核心优势：Spark会把任务拆分到多个节点并行处理，避免单节点内存和性能瓶颈；所有操作都是分布式向量化的，处理大表效率远超Pandas。

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