如何用Pandas按特定规则匹配两个CSV文件的行？

当前需求是给两个CSV文件的行做最优匹配（按自定义分数规则筛选最高分匹配项），原嵌套循环实现在大数据量下效率极低，希望用Pandas重写逻辑提升性能，且已有明确的SQL匹配规则，以下是最适用的算法和实现思路：

一、预过滤+向量化计算（最实用的基础方案） #

• 核心逻辑：先通过高权重字段（ISRC、UPC）做精准匹配缩减候选集，避免全量笛卡尔积导致的内存溢出，再用Pandas/Numpy的向量化操作批量计算分数，彻底抛弃低效循环。
• 操作步骤：
  1. 数据预处理：将两个文件的文本字段统一转大写，把空值、空字符串替换为统一标识（比如空串''），重命名字段对齐匹配规则。
  2. 分层匹配：先对ISRC、UPC这类高权重字段执行merge精准匹配，筛选出已匹配的行；对未匹配的行，通过文本前缀（比如标题前3个字符）做merge缩小候选集，避免全量对比。
  3. 向量化算分：用np.select或np.where替代逐行判断，批量计算标题、艺术家、专辑等字段的匹配分数，最后求和得到总分。
  4. 筛选最优匹配：对每个A行的候选匹配结果按总分降序排序，保留第一行（最高分），对无匹配的行标记Unmatched。

二、分组候选集缩减（针对超大规模文本数据集） #

• 核心逻辑：对文本字段（标题、艺术家、专辑）提取特征（比如前缀、分词首词），按特征分组，让每个A行仅与同组的B行做匹配，大幅降低计算量。
• 操作示例：给B的标题生成title_prefix字段（取前3个大写字符），A的标题也生成相同前缀，仅merge前缀相同的行，候选集规模可缩小90%以上，适合十万级及以上的数据量。

三、模拟SQL窗口逻辑（快速对齐现有需求） #

• 核心逻辑：直接将你编写的SQL逻辑翻译成Pandas代码，利用分组排序取最优的方式，和SQL的ROW_NUMBER()逻辑完全对齐。
• 关键优化：若数据量较大，禁止直接执行全量cross join（会导致内存爆炸），必须先做预过滤；用sort_values+drop_duplicates替代groupby.apply，后者在大数据场景下效率极低。

四、近似字符串匹配（提升文本匹配精度） #

• 核心逻辑：如果startswith的匹配规则不够精准，可引入近似字符串匹配算法（比如编辑距离/Levenshtein距离），结合权重计算总分。
• 工具与注意事项：使用rapidfuzz库（比fuzzywuzzy性能提升数倍），但需先通过前缀/精准匹配筛选候选集，再对候选集计算近似匹配分数，避免全量计算的性能损耗。

Pandas实现示例代码 #

import pandas as pd
import numpy as np

# 读取CSV文件
df_a = pd.read_csv('fileA.csv')  # 对应SQL中的blocklist表
df_b = pd.read_csv('fileB.csv')  # 对应SQL中的source_catalog表

# 数据预处理：统一格式、重命名字段
df_a = df_a.assign(
    b_title=df_a['title'].str.upper().fillna(''),
    b_artist=df_a['artist'].str.upper().fillna(''),
    b_album=df_a['album'].str.upper().fillna(''),
    b_isrc=df_a['isrc'].fillna(''),
    b_upc=df_a['upc'].fillna(0)
).rename(columns={'id': 'b_id'})

df_b = df_b.assign(
    sc_title=df_b['track_title'].str.upper().fillna(''),
    sc_artist=df_b['track_artist'].str.upper().fillna(''),
    sc_album=df_b['album_title'].str.upper().fillna(''),
    sc_isrc=df_b['isrc'].fillna(''),
    sc_upc=df_b['upc'].fillna(0)
).rename(columns={'id': 'sc_id'})

# 分层匹配缩减候选集
# 1. ISRC精准匹配
matched_isrc = pd.merge(df_a, df_b, left_on='b_isrc', right_on='sc_isrc', how='inner')
# 2. UPC精准匹配（排除已通过ISRC匹配的行）
unmatched_a = df_a[~df_a['b_id'].isin(matched_isrc['b_id'])]
matched_upc = pd.merge(unmatched_a, df_b, left_on='b_upc', right_on='sc_upc', how='inner')
# 3. 前缀模糊匹配（剩余未匹配行）
remaining_a = df_a[~df_a['b_id'].isin(pd.concat([matched_isrc['b_id'], matched_upc['b_id']]))]
remaining_a['title_prefix'] = remaining_a['b_title'].str[:3]
df_b['title_prefix'] = df_b['sc_title'].str[:3]
matched_fuzzy = pd.merge(remaining_a, df_b, on='title_prefix', how='inner')

# 合并所有候选集
all_candidates = pd.concat([matched_isrc, matched_upc, matched_fuzzy], ignore_index=True)

# 向量化计算各维度分数
# 标题分数
all_candidates['titles_score'] = np.select(
    [
        (all_candidates['b_title'] == all_candidates['sc_title']) & (all_candidates['b_title'] != ''),
        (all_candidates['b_title'].str.startswith(all_candidates['sc_title']) | all_candidates['sc_title'].str.startswith(all_candidates['b_title'])) & (all_candidates['b_title'] != '') & (all_candidates['sc_title'] != '')
    ],
    [110, 100],
    default=0
)

# 艺术家分数
all_candidates['artists_score'] = np.select(
    [
        (all_candidates['b_artist'] == all_candidates['sc_artist']) & (all_candidates['b_artist'] != ''),
        (all_candidates['b_artist'].str.startswith(all_candidates['sc_artist']) | all_candidates['sc_artist'].str.startswith(all_candidates['b_artist'])) & (all_candidates['b_artist'] != '') & (all_candidates['sc_artist'] != '')
    ],
    [35, 30],
    default=0
)

# 专辑分数
all_candidates['albums_score'] = np.select(
    [
        (all_candidates['b_album'] == all_candidates['sc_album']) & (all_candidates['b_album'] != ''),
        (all_candidates['b_album'].str.startswith(all_candidates['sc_album']) | all_candidates['sc_album'].str.startswith(all_candidates['b_album'])) & (all_candidates['b_album'] != '') & (all_candidates['sc_album'] != '')
    ],
    [25, 20],
    default=0
)

# ISRC分数
all_candidates['isrcs_score'] = np.where(
    (all_candidates['b_isrc'] == all_candidates['sc_isrc']) & (all_candidates['b_isrc'] != ''),
    500, 0
)

# UPC分数
all_candidates['upcs_score'] = np.where(
    (all_candidates['b_upc'] == all_candidates['sc_upc']) & (all_candidates['b_upc'] != 0),
    100, 0
)

# 计算总分
all_candidates['total_score'] = all_candidates[['titles_score', 'artists_score', 'albums_score', 'isrcs_score', 'upcs_score']].sum(axis=1)

# 筛选每个b_id的最高分匹配项
all_candidates_sorted = all_candidates.sort_values(['b_id', 'total_score'], ascending=[True, False])
best_matches = all_candidates_sorted.drop_duplicates(subset='b_id', keep='first')

# 处理未匹配的行
unmatched_final = df_a[~df_a['b_id'].isin(best_matches['b_id'])].assign(
    matched='Unmatched',
    total_score=0
)

# 合并最终结果
final_result = pd.concat([
    best_matches[['b_id', 'sc_id', 'total_score']].rename(columns={'sc_id': 'matched'}),
    unmatched_final[['b_id', 'matched', 'total_score']]
], ignore_index=True)

# 关联原表字段（若需要保留原行的所有信息）
final_result = pd.merge(df_a, final_result, on='b_id', how='left')

# 输出结果
final_result.to_csv('matched_result.csv', index=False)

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