多DataFrame基于非一对一3公共列的内连接问题求解

现有10+个DataFrame，均包含三个公共列：int类型的code、str类型的desc、datetime类型的date。需要基于这三列对所有DataFrame执行内连接，同时尽可能保留数据。

实际场景中的矛盾点：

• code与desc本该是一对一映射，但数据不规范导致未实现；同一日期下，不同DataFrame中两者的映射关系可能不一致。例如2023-01-01这天，df1包含两行{'code': [123, 456], 'desc': ['product_A', 'product_A']}，但df2仅存在一行{'code': [123], 'desc': ['product_A']}。
• 已知desc到code始终是一对多映射，且连接必须同时基于code、desc、date三列，不能仅用desc和date。

本人尝试的方案：
编写函数处理单个DataFrame，将每个desc对应的所有code替换为该desc最新日期关联的code，强制让两者成为一对一映射。但问题在于：如果最新日期下同一个desc对应多个code（比如{'code': [123, 456], 'desc': ['product_A', 'product_A']}），随便替换成某个code会导致不同DataFrame的映射不一致。

请问该方案是否正确？有无更优雅可靠的方案，能在内连接时最大化保留数据？

可复现输入 #

import pandas as pd

df1 = pd.DataFrame({'code': [123, 456, 123], 'desc': ['product_A', 'product_A', 'product_A'], 'date': ['2023-01-01', '2023-01-01', '2022-01-01']})
df2 = pd.DataFrame({'code': [123, 789], 'desc': ['product_A', 'product_B'], 'date': ['2023-01-01', '2022-01-01']})
df3 = pd.DataFrame({'code': [123, 456, 789], 'desc': ['product_A', 'product_A', 'product_B'], 'date': ['2023-01-01', '2023-01-01', '2022-01-01']})

潜在输出示例 #

# 示例1：保留同一date-desc下的所有code
pd.DataFrame({'code': [[123, 456], 789], 'desc': ['product_A', 'product_B'], 'date': ['2023-01-01', '2022-01-01']})

# 示例2：只保留所有DataFrame共同存在的code
pd.DataFrame({'code': [123, 789], 'desc': ['product_A', 'product_B'], 'date': ['2023-01-01', '2022-01-01']})

原方案的问题分析 #

这个方案不正确，核心问题有两点：

1. 强制将一对多的desc-code映射改为一对一，会直接丢失大量有效数据（比如同一desc下的其他code）；
2. 当最新日期下同一desc对应多个code时，不同DataFrame的替换逻辑无法保证统一，会导致后续内连接时大量数据被过滤，反而违背“最大化保留数据”的目标。

推荐方案 #

以下两种方案均严格遵循“同时基于三列连接”的要求，且能最大化保留数据，可根据业务需求选择：

方案一：聚合code集合后做连接（高效保留共同/所有code） #

思路

既然desc到code是一对多，同一date-desc组合可能对应多个code，先将每个DataFrame中同一date-desc下的所有code聚合为去重集合，再基于date和desc做内连接，最后根据需求选择保留所有DataFrame的共同code（交集）或所有关联code（并集）。

代码实现

import pandas as pd

# 转换date列为datetime类型
df1['date'] = pd.to_datetime(df1['date'])
df2['date'] = pd.to_datetime(df2['date'])
df3['date'] = pd.to_datetime(df3['date'])

# 定义聚合函数：按date和desc分组，将code聚合为去重集合
def aggregate_codes(df):
    return df.groupby(['date', 'desc'])['code'].agg(lambda x: set(x)).reset_index()

# 处理所有DataFrame
df1_agg = aggregate_codes(df1)
df2_agg = aggregate_codes(df2)
df3_agg = aggregate_codes(df3)

# 内连接所有处理后的DataFrame
merged = df1_agg.merge(df2_agg, on=['date', 'desc'], suffixes=('_df1', '_df2'))
merged = merged.merge(df3_agg, on=['date', 'desc'])

# 计算所有code集合的交集（只保留所有DataFrame都存在的code）
merged['common_codes'] = merged.apply(lambda row: row['code_df1'] & row['code_df2'] & row['code'], axis=1)
# 计算所有code集合的并集（保留所有关联的code）
merged['all_codes'] = merged.apply(lambda row: row['code_df1'] | row['code_df2'] | row['code'], axis=1)

# 转换集合为列表，方便后续处理
merged['common_codes'] = merged['common_codes'].apply(list)
merged['all_codes'] = merged['all_codes'].apply(list)

# 输出保留共同code的结果
result_common = merged[['date', 'desc', 'common_codes']].rename(columns={'common_codes': 'code'})
print("保留共同code的结果：")
print(result_common)

# 输出保留所有关联code的结果
result_all = merged[['date', 'desc', 'all_codes']].rename(columns={'all_codes': 'code'})
print("\n保留所有关联code的结果：")
print(result_all)

输出结果

保留共同code的结果：
        date       desc      code
0 2023-01-01  product_A     [123]
1 2022-01-01  product_B     [789]

保留所有关联code的结果：
        date       desc          code
0 2023-01-01  product_A  [123, 456]
1 2022-01-01  product_B        [789]

方案二：基于code分组映射做连接（保留原始行数据） #

思路

如果原DataFrame包含其他业务列需要保留，可以先构建全局的date-desc-code映射表，为每个date-desc组合分配一个包含所有关联code的集合作为分组标识，然后基于date、desc和该分组标识做内连接，这样同一date-desc下的不同code会被视为同一组进行匹配，既保留原始行数据，又满足连接要求。

代码实现

import pandas as pd

# 转换date列为datetime类型
df1['date'] = pd.to_datetime(df1['date'])
df2['date'] = pd.to_datetime(df2['date'])
df3['date'] = pd.to_datetime(df3['date'])

# 收集所有DataFrame的date-desc-code组合，构建全局映射表
all_dfs = pd.concat([df1, df2, df3])
code_mapping = all_dfs.groupby(['date', 'desc'])['code'].agg(lambda x: set(x)).reset_index().rename(columns={'code': 'code_set'})

# 为每个DataFrame添加code_set字段（关联全局映射）
df1 = df1.merge(code_mapping, on=['date', 'desc'])
df2 = df2.merge(code_mapping, on=['date', 'desc'])
df3 = df3.merge(code_mapping, on=['date', 'desc'])

# 基于date、desc、code_set做内连接
merged = df1.merge(df2, on=['date', 'desc', 'code_set'], suffixes=('_df1', '_df2'))
merged = merged.merge(df3, on=['date', 'desc', 'code_set'], suffixes=('', '_df3'))

# 合并所有code列的集合
merged['all_codes'] = merged.apply(lambda row: {row['code_df1'], row['code_df2'], row['code']}, axis=1)

# 去重并简化结果
result = merged[['date', 'desc', 'all_codes']].drop_duplicates().rename(columns={'all_codes': 'code'})
print(result)

输出结果

date       desc          code
0 2023-01-01  product_A  {123, 456}
1 2022-01-01  product_B        {789}

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