Pandas自定义transform去重失效 如何获取输入文件生成DataFrame

1. transform函数获取测试文件DataFrame的逻辑 #

你所使用的基于BaseTransformer的ETL框架已经封装了前置的文件读取逻辑：测试运行时，框架会自动定位对应测试用例的输入文件，按照配置的格式规则（CSV/Parquet/JSON等）将文件全量内容解析为Pandas DataFrame对象，作为入参自动传入你重写的transform(self, df)方法。你不需要手动编写文件读取代码，框架已经完成了IO层的处理。
你提到的distinct().py是脚本文件名，框架会按照约定自动扫描脚本内继承了BaseTransformer的实现类，加载数据集后直接调用其transform方法，不需要你手动传参。

2. 去重未生效的原因与修复方法 #

你的代码存在冗余逻辑和参数缺失问题，具体如下：

• 冗余逻辑：iterrows()逐行遍历完全没有必要，drop_duplicates()本身是全表级别的操作，你在循环内重复两次对全表执行去重、重复套pd.DataFrame()构造对象，不仅执行效率极低，遍历过程中修改原DataFrame还可能触发隐式的索引对齐问题。
• 核心原因：drop_duplicates()默认采用全列值完全匹配的规则判定重复行，如果测试数据中你认为的“重复行”存在任意列值差异（比如自增主键、入库时间戳、隐藏索引列值不同），就不会被判定为重复，最终返回结果和输入完全一致。

修复后代码 #

from transformer import BaseTransformer
import pandas as pd

class MyTransformer(BaseTransformer):
    def __init__(self):
        super().__init__()

    def transform(self, df):
        # 排查时可先放开下面这行打印，确认默认规则下的重复行数量
        # print(f"默认全列匹配下重复行数：{df.duplicated().sum()}")
        
        # 全列去重使用这行
        # df_dedup = df.drop_duplicates(keep="first")
        
        # 按指定业务字段去重使用这行，把subset里的内容替换成你判定重复的实际列名
        df_dedup = df.drop_duplicates(subset=["业务主键1", "业务主键2"], keep="first")
        return df_dedup

排查提示 #

如果打印出的默认全列匹配重复行数为0，说明你需要通过subset参数明确指定判定重复的业务字段，不要使用默认的全列匹配规则。

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