数据清洗中替代iterrows()的高效方案？基于fuzzywuzzy的X列拼写纠正

问题描述 #

我正在编写数据清洗脚本，处理数据表X列的拼写错误问题，使用fuzzywuzzy进行模糊字符串匹配来识别并纠正错误。X列有效值约50种，当Y列值非空时，可通过Y列（X与Y存在函数依赖）将X候选有效值缩小至约5种。我使用单独的correct_values数据表存储X列正确拼写值及对应Y列值，当前基于iterrows()的方案可行但效率较低，特此询问是否有更高效的实现方式？

原实现代码 #

def cleanup_misspelled_X():

        # Checks for rows where X is invalid and not null
        # These rows's X value could be misspelled
        if (working_df[
                        (~working_df['X'].isin(correct_values['X'])) &
                        (~working_df['X'].isna())
                      ]).any:

            # If so, check if those same rows have a valid/present Y value.
            # If the row has a valid/present Y value, we can narrow down the
            # possible X values
            if (working_df[
                            (~working_df['X'].isin(correct_values['X'])) &
                            (~working_df['X'].isna()) &
                            (working_df['Y'].isin(correct_values['Y']))
                           ]).any:

                # If so, assign those records to a dataframe
                to_be_corrected_df = working_df[
                            (~working_df['X'].isin(correct_values['X'])) &
                            (~working_df['X'].isna()) &
                            (working_df['Y'].isin(correct_values['Y']))
                                                ]

                # Iterate through those rows
                for index, row in to_be_corrected_df.iterrows():

                    # If match is > 85%, fill the invalid X value with the match
                    if process.extractOne(
                           # value to attempt to match
                           row['X'],
                            # collection of values to match against 
                           correct_values.loc[correct_values['Y'] == row['Y']]['X']
                                         )[1] > 85:

                        working_df.loc[
                           working_df['row_identifier'] == row['row_identifier'],
                           ['X']] = \

                           process.extractOne(
                           row['X'],
                           correct_values.loc[correct_values['Y'] == row['Y']]['X']
                                             )[0]

                    else:
                        pass # for now 

            # If the row's X value is invalid and not null
            # and said row does not have a valid/present Y value
            if (working_df[
                            (~working_df['X'].isin(correct_values['X'])) &
                            (~working_df['X'].isna()) &
                            (~working_df['Y'].isin(correct_values['Y']))
                           ]).any:

                # Assign such rows to a dataframe
                to_be_corrected_df = working_df[
                            (~working_df['X'].isin(correct_values['X'])) &
                            (~working_df['X'].isna()) &
                            (~working_df['Y'].isin(correct_values['Y']))
                                                ]

                # Iterate through the dataframe
                for index, row in to_be_corrected_df.iterrows():

                    # If the proposed match is > 90% then fill the invalid X value 
                    # with the match
                    if process.extractOne(
                                        row['X'], 
                                        correct_values['X']
                                        )[1] > 90:

                        working_df.loc[
                            working_df['row_identifier'] == row['row_identifier'], 
                            ['X']] = \

                            process.extractOne(
                                        row['X'], 
                                        correct_values['X']
                                        )[0]

                    else:
                        pass # for now

高效优化方案 #

原方案的核心效率瓶颈在于iterrows()的纯Python逐行循环、重复的DataFrame切片查询，以及重复调用process.extractOne()。以下是针对性的优化实现：

优化思路 #

• 提前预处理映射：将correct_values按Y分组生成字典，避免每次循环执行loc查询
• 用apply替代iterrows：apply基于C级循环，比纯Python循环效率提升数倍
• 避免重复计算：每个待处理行仅调用一次extractOne()，同时获取匹配值和分数
• 批量标记与更新：先标记所有待处理行，再统一批量更新，减少DataFrame操作次数

优化后代码 #

from fuzzywuzzy import process

# 提前预处理correct_values，生成映射字典和全局有效值列表（只需执行一次）
y_to_valid_x = correct_values.groupby('Y')['X'].apply(list).to_dict()
all_valid_x = correct_values['X'].tolist()

def cleanup_misspelled_X_optimized():
    # 标记所有X无效且非空的行
    mask_invalid_x = (~working_df['X'].isin(all_valid_x)) & (~working_df['X'].isna())
    if not mask_invalid_x.any():
        return  # 无待处理行，直接返回
    
    # 拆分两种场景：Y值有效、Y值无效
    mask_y_valid = mask_invalid_x & working_df['Y'].isin(y_to_valid_x.keys())
    mask_y_invalid = mask_invalid_x & ~working_df['Y'].isin(y_to_valid_x.keys())
    
    # 处理Y值有效的场景：基于Y缩小候选X范围
    if mask_y_valid.any():
        def match_x_with_y(row):
            # 直接从预构建的字典中获取候选X列表
            candidate_x = y_to_valid_x[row['Y']]
            match_str, match_score = process.extractOne(row['X'], candidate_x)
            # 分数达标则替换，否则保留原X值
            return match_str if match_score > 85 else row['X']
        
        # 批量更新X列
        working_df.loc[mask_y_valid, 'X'] = working_df[mask_y_valid].apply(match_x_with_y, axis=1)
    
    # 处理Y值无效的场景：匹配全局所有X有效值
    if mask_y_invalid.any():
        def match_x_global(row):
            match_str, match_score = process.extractOne(row['X'], all_valid_x)
            return match_str if match_score > 90 else row['X']
        
        # 批量更新X列
        working_df.loc[mask_y_invalid, 'X'] = working_df[mask_y_invalid].apply(match_x_global, axis=1)

额外优化建议 #

如果数据量极大，还可以考虑：

• 用rapidfuzz替代fuzzywuzzy：rapidfuzz是fuzzywuzzy的C++重写版本，匹配速度提升10-100倍，API完全兼容
• 并行处理：对大批次的待处理行使用多进程/多线程并行执行匹配，进一步提升效率

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