如何在两个CSV文件中找出所有主键与外键候选列集及匹配方案？

现有两个非规范化CSV文件PAYMENT和CUSTOMER（实际最多各含100列），示例数据如下：

PAYMENT表示例 #

ID, CUST_NAME, CUST_NUM, CLIENT_NAME, PAYMENT_NUM, START_DATE,  END_DATE
1,     CUST1,   A001,    CLIENT1,       10,        2018-04-01, 2018-04-02
2,     CUST1,   A001,    CLIENT1,       10,        2018-04-01, 2018-05-30
3,     CUST2,   A002,    CLIENT1,      101,        2018-04-02, 2018-04-03
4,     CUST2,   A002,    CLIENT1,      102,        2018-04-02, 2018-04-03

CUSTOMER表示例 #

ID, CUST_NAME, CUST_NUM, AGE, GENDER, COUNTRY
1,     CUST1,   A001,     32,   M,       US
2,     CUST2,   A002,     34,   F,       CA
3,     CUST3,   A003,     45,   M,       US
4,     CUST4,   A004,     31,   F,       CA

需求是找出所有可能的主键（PK）与外键（FK）候选列集，期望输出如下：

1. CUSTOMER.CUST_NAME (PK), PAYMENT.CUST_NAME (FK)
2. CUSTOMER.CUST_NUM (PK), PAYMENT.CUST_NUM (FK)
3. CUSTOMER.CUST_NAME (PK), CUSTOMER.CUST_NUM (PK), PAYMENT.CUST_NAME (FK), PAYMENT.CUST_NUM (FK)

目前已通过pandas和itertools实现主键候选的查找，需要进一步实现外键候选的查找及匹配逻辑。

步骤1：明确主键候选集 #

先确认你已通过以下逻辑得到CUSTOMER表的所有主键候选（单列或多列组合）：

• 单列候选：检查列内无重复值，即df[col].nunique() == len(df)
• 多列候选：用itertools.combinations生成所有列组合，验证组合后的行是否唯一，即df.groupby(comb).size().max() == 1

将CUSTOMER的PK候选保存为集合，比如customer_pk_candidates，每个元素是列名元组（如('CUST_NAME',)、('CUST_NUM',)、('CUST_NAME', 'CUST_NUM')）。

步骤2：匹配列对并验证外键约束 #

对于每个CUSTOMER的PK候选，在PAYMENT中找到对应同名列组合，验证外键规则：PAYMENT中该列组合的所有非空值，必须全部存在于CUSTOMER对应列组合的值集合中（可根据业务需求调整是否允许空值）。

示例代码实现：

import pandas as pd
import itertools

# 读取数据
payment_df = pd.read_csv('PAYMENT.csv')
customer_df = pd.read_csv('CUSTOMER.csv')

# 1. 获取CUSTOMER的所有PK候选
customer_pk_candidates = []
cols = customer_df.columns.tolist()

# 单列PK候选
for col in cols:
    if customer_df[col].nunique() == len(customer_df):
        customer_pk_candidates.append((col,))

# 多列PK候选（可调整组合列数上限）
for k in range(2, len(cols)+1):
    for comb in itertools.combinations(cols, k):
        if customer_df.groupby(list(comb)).size().max() == 1:
            customer_pk_candidates.append(comb)

# 2. 筛选并验证有效FK候选
valid_fk_pairs = []
for pk_comb in customer_pk_candidates:
    # 检查PAYMENT是否有对应列
    if all(col in payment_df.columns for col in pk_comb):
        # 提取CUSTOMER的PK值集合
        customer_pk_set = set(customer_df[list(pk_comb)].apply(tuple, axis=1))
        # 提取PAYMENT的候选FK值（排除空值行）
        payment_fk_vals = payment_df[list(pk_comb)].dropna().apply(tuple, axis=1)
        # 验证所有FK值都在PK集合内
        if all(val in customer_pk_set for val in payment_fk_vals):
            # 格式化输出字符串
            pk_str = ', '.join([f'CUSTOMER.{col} (PK)' for col in pk_comb])
            fk_str = ', '.join([f'PAYMENT.{col} (FK)' for col in pk_comb])
            valid_fk_pairs.append(f'{pk_str}, {fk_str}')

# 输出结果
for idx, pair in enumerate(valid_fk_pairs, 1):
    print(f'{idx}. {pair}')

步骤3：输出结果 #

运行上述代码后，会得到与期望一致的PK-FK候选列表，自动包含单列和多列的有效组合。

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