解决Pandas中apply返回DataFrame后合并报错的问题

问题根源 #

1. 使用apply(axis=1)调用返回DataFrame的函数时，得到的是元素为DataFrame的Series，Series本身没有merge方法，因此触发Series object has no attribute 'merge'错误。
2. 将该Series转为DataFrame后，列中存储的是DataFrame对象，合并时Pandas会尝试将这些对象作为哈希键处理，但DataFrame属于不可哈希类型，因此触发unhashable type: 'DataFrame'错误。

修复步骤 #

1. 重构函数调用逻辑，合并小DataFrame为统一大表 #

放弃用apply直接存储结果，改为遍历df_src1的每一行，收集所有聚合后的DataFrame，再合并成一个标准的DataFrame：

# 初始化列表存储所有聚合结果
agg_dfs = []

# 遍历df_src1的每一行，调用聚合函数并收集结果
for idx, row in df_src1.iterrows():
    col_name = row['column_name']
    table_name = row['schema_table']
    # 调用自定义聚合函数
    df_agg = src_aggregated_input(col_name, table_name)
    # 添加来源表和列的标识列，方便后续合并关联
    df_agg['source_table'] = table_name
    df_agg['source_column'] = col_name
    agg_dfs.append(df_agg)

# 合并所有小DataFrame为一个大的标准DataFrame
df_src2 = pd.concat(agg_dfs, ignore_index=True)

2. 修复自定义函数的潜在问题 #

原函数中直接修改全局的df_src1会导致逻辑污染，同时异常返回np.nan会干扰后续合并，调整如下：

def src_aggregated_input(columnname, tablename):
    try:
        # 检查传入的列名是否为空，替代原全局修改逻辑
        if pd.isna(columnname) or columnname.strip() == '':
            return pd.DataFrame(columns=['src_agg_cnt'])
        
        # 使用f-string简化SQL拼接，避免字符串拼接错误
        sql = f'SELECT COUNT({columnname}) AS src_agg_cnt, {columnname} FROM {tablename} GROUP BY {columnname}'
        loging(datetime.datetime.now(), '-agg src sql', sql)
        
        df_agg_src = pd.read_sql_query(sql, db_conn)
        print("df_agg_src", df_agg_src)
        print("src", type(df_agg_src))
        return df_agg_src

    except Exception as e:
        exc_type, exc_obj, exc_tb = sys.exc_info()
        fname = os.path.split(exc_tb.tb_frame.f_code.co_filename)[1]
        loging(datetime.datetime.now(), 'src_agg', sys.exc_info()[1])
        # 异常时返回结构匹配的空DataFrame，避免concat出错
        return pd.DataFrame(columns=['src_agg_cnt', columnname])

3. 正常执行合并操作 #

现在df_src2是标准的DataFrame，可直接与df_tgt2执行合并：

def aggregated_cnt_check():
    try:
        # 根据实际业务逻辑调整关联键，示例用来源表、来源列、聚合字段关联
        df_agg_merge = df_src2.merge(
            df_tgt2,
            on=['source_table', 'source_column', df_src2.columns[1]],  # 替换为实际关联列
            indicator=True,
            how='outer'
        )
        # 筛选出两边存在差异的记录
        df_agg_diff = df_agg_merge[df_agg_merge['_merge'] != 'both']
        return df_agg_diff
    except Exception as e:
        loging(datetime.datetime.now(), 'merge error', str(e))
        raise

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