基于行值模糊匹配合并多DataFrame时列丢失问题排查

问题场景 #

需要合并n个DataFrame，每个表都包含公司名列（名称可能不完全一致），部分列（如Value、Currency）在各表间一致，部分列则为子表独有。要求基于公司名称做模糊匹配合并，同一列的不同值以元组形式存储，但运行代码后发现合并结果未包含所有子表的列。

原代码 #

def concatenate_tables(self, tables):
    def normalize_name(name):
        if pd.isna(name) or not isinstance(name, str) or name.strip() == '':
            return None
        return name.lower().strip()

    def fuzzymatch(name, existing_entities): # Uses fuzzywuzzy
        if not name or not existing_entities:
            return None
        matches = process.extract(name, existing_entities, scorer=fuzz.token_set_ratio, limit=1)
        return matches[0][0] if matches and matches[0][1] > 80 else None

    # Ensure all inputs are DataFrames
    if not all(isinstance(df, pd.DataFrame) for df in tables):
        raise ValueError("All elements in 'tables' must be pandas DataFrames.")

    # Collect all unique columns and company names
    all_columns = set(['Company Name'])  # Initialize with 'Company Name' to ensure inclusion
    for df in tables:
        all_columns.update(df.columns)

    all_company_names = set()
    for df in tables:
        if 'Company Name' in df.columns:
            all_company_names.update(df['Company Name'].astype(str).apply(normalize_name).dropna())

    consolidated_data = defaultdict(lambda: {col: [] for col in all_columns})  # Pre-fill with all columns
    for df in tables:
        for _, row in df.iterrows():
            raw_name = row['Company Name']
            norm_name = normalize_name(raw_name)
            matched_name = fuzzymatch(norm_name, all_company_names) if norm_name else None
            effective_name = matched_name if matched_name else norm_name  # Use normalized name if no match

            for col in all_columns:
                # Append data if column exists in current df, else append None
                if col in df.columns:
                    consolidated_data[effective_name][col].append(row[col])
                else:
                    consolidated_data[effective_name][col].append(None)

    # Prepare consolidated rows
    consolidated_rows = []
    for company_name, cols in consolidated_data.items():
        row_data = {"Company Name": company_name}
        for col, values in cols.items():
            # Filter out None values
            non_null_values = list(filter(None.__ne__, values))
            if len(set(non_null_values)) > 1:
                row_data[col] = tuple(set(non_null_values))
            elif non_null_values:
                row_data[col] = non_null_values[0]
            else:
                row_data[col] = None

        consolidated_rows.append(row_data)

    master_df = pd.DataFrame(consolidated_rows)

    return master_df

排查原因及修复方案 #

1. 列名未标准化导致重复/丢失 #

原代码直接用df.columns收集列名，若子表列名存在大小写、空格差异（如Value和value、Company Name和CompanyName），会被识别为不同列，或因后续处理逻辑遗漏。

修复：添加列名标准化逻辑，统一格式：

def normalize_col(col):
    if isinstance(col, str):
        # 转小写、替换空格为下划线、去除首尾空格
        return col.strip().lower().replace(' ', '_')
    return str(col)

# 收集列名时先标准化
all_columns = {normalize_col('Company Name')}
for df in tables:
    # 重命名子表列名统一格式
    df.rename(columns=lambda x: normalize_col(x), inplace=True)
    all_columns.update(df.columns)

# 后续代码中统一使用标准化后的公司名列名（如company_name）

2. 硬编码公司名列名，未适配子表的非标准列名 #

原代码假设所有子表的公司名都叫Company Name，但实际子表可能用Corp Name、企业名称等列名，导致该表的列未被正确纳入all_columns，甚至抛出KeyError。

修复：自动检测并统一公司名列名：

def detect_company_col(df):
    # 优先匹配含company和name的列
    possible_cols = [col for col in df.columns 
                    if 'company' in col and 'name' in col]
    if possible_cols:
        return possible_cols[0]
    # 备选匹配含corp、firm等关键词的列
    possible_cols = [col for col in df.columns 
                    if any(keyword in col for keyword in ['corp', 'firm', 'enterprise'])]
    return possible_cols[0] if possible_cols else None

# 预处理子表，统一公司名列名
for df in tables:
    company_col = detect_company_col(df)
    if company_col:
        df.rename(columns={company_col: 'company_name'}, inplace=True)

3. 公司名列被重复覆盖导致逻辑异常 #

原代码中row_data先设置"Company Name": company_name，后续循环cols.items()时又会处理Company Name列，覆盖之前设置的标准化名称，同时可能干扰其他列的处理。

修复：跳过公司名列的循环处理：

# 构建合并行时跳过公司名列
row_data = {"company_name": company_name}
for col, values in cols.items():
    if col == "company_name":
        continue  # 保留手动设置的标准化公司名，跳过列值合并逻辑
    non_null_values = list(filter(None.__ne__, values))
    if len(set(non_null_values)) > 1:
        row_data[col] = tuple(set(non_null_values))
    elif non_null_values:
        row_data[col] = non_null_values[0]
    else:
        row_data[col] = None

4. 集合无序导致列初始化遗漏（潜在风险） #

原代码用集合存储all_columns，虽然不影响列的存在，但集合无序可能导致consolidated_data中列的顺序混乱，若后续依赖列顺序会出问题。

修复：将集合转为有序列表：

all_columns = sorted(all_columns)
consolidated_data = defaultdict(lambda: {col: [] for col in all_columns})

验证步骤 #

1. 打印all_columns，确认包含所有子表的列（标准化后）。
2. 检查每个子表的列名是否已统一为标准化格式。
3. 查看consolidated_data中任意公司的字典，确认包含所有列。
4. 输出master_df.columns，验证是否覆盖所有子表列。

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