如何参照df1层级规则修正df2分类字段并简化Pandas合并代码

df1为标准层级映射表，存储了small_class→big_class→category的正确对应关系，示例数据如下：

category  category_id big_class  big_class_id small_class  small_class_id
0        A            1        a1            11         aa1             111
1        A            1        a2            12         aa2             112
2        B            2        b1            21         bb1             221
3        B            2        b2            22         bb2             222
4        B            2        b3            23         bb3             223

df2为待修正的业务数据表，其中层级字段存在错误，需要以df1的映射关系为标准，按照细粒度优先的规则修正所有层级字段，匹配优先级：small_class > big_class > category。df2示例数据如下：

item small_class  small_class_id  ... big_class_id  category category_id
0  aaa1         aa1           111.0  ...         12.0         B         2.0
1  aaa2         aa2           112.0  ...         11.0       NaN         NaN
2  bbb3         NaN             NaN  ...          NaN         B         2.0
3  bbb4         bb3           223.0  ...          NaN         A         1.0
4  bbb1         NaN             NaN  ...         23.0         A         1.0

最终预期输出：

item small_class  small_class_id  ... big_class_id     category category_id
0  aaa1         aa1           111.0  ...         11.0            A           1
1  aaa2         aa2           112.0  ...         12.0            A           1
2  bbb3         NaN             NaN  ...          NaN            B           2
3  bbb4         bb3           223.0  ...         23.0            B           2
4  bbb1         NaN             NaN  ...         22.0            B           2

下面两种方案都比原实现逻辑更清晰、代码更简洁：

方案1：左连+空值填充（性能更优，适合大数据量） #

# 第一步：以small_class为键关联标准表，匹配到的字段直接覆盖原值
tmp = df2.merge(df1, on="small_class", how="left", suffixes=("_old", ""))

# 第二步：剩余未匹配到的行，以big_class为键关联标准表上层字段
bc_ref = df1.drop_duplicates("big_class")[["big_class", "big_class_id", "category", "category_id"]]
tmp = tmp.merge(bc_ref, on="big_class", how="left", suffixes=("", "_bc"))

# 第三步：按优先级填充空值，细粒度匹配结果>粗粒度匹配结果>原值
fix_cols = ["category", "category_id", "big_class_id", "small_class_id"]
for col in fix_cols:
    tmp[col] = tmp[col].fillna(tmp[f"{col}_bc"]).fillna(tmp[f"{col}_old"])

# 输出结果保留所需字段
res = tmp[["item", "small_class", "small_class_id", "big_class", "big_class_id", "category", "category_id"]]

方案2：行级自定义函数（逻辑更直观，适合小数据量、规则灵活调整场景） #

import pandas as pd

# 提前构建不同粒度的映射字典
sc_map = df1.set_index("small_class").to_dict("index")
bc_map = df1.drop_duplicates("big_class").set_index("big_class")[["big_class_id", "category", "category_id"]].to_dict("index")

def fix_level(row):
    # 优先用small_class匹配全量层级
    if pd.notna(row["small_class"]) and row["small_class"] in sc_map:
        return pd.Series(sc_map[row["small_class"]])
    # 匹配失败再用big_class匹配上层层级
    if pd.notna(row["big_class"]) and row["big_class"] in bc_map:
        return pd.Series({**row.to_dict(), **bc_map[row["big_class"]]})
    # 都匹配失败保留原值
    return row

res = df2.apply(fix_level, axis=1)

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