Python/Pandas处理一对多CSV合并以实现反规范化的问题

嘿，我完全懂你遇到的困扰——当处理带一对多关系的CSV合并时，一不小心就会出现记录被覆盖的情况，其实核心是没搞清楚Pandas在处理这种关联时的默认行为，以及如何正确应对。我来给你一步步拆解解决方案：

问题根源先搞懂 #

首先要明确：Pandas的merge函数在遇到一对多关系时，默认不会覆盖值，而是会把所有匹配的记录都展开成多行。如果你的代码出现了覆盖，大概率是后续的处理逻辑（比如不小心用了drop_duplicates、或者错误地指定了合并类型）导致的，先检查这部分哦。

基础解决方案：保留所有匹配记录 #

先从最基础的正确合并流程说起，假设你的三个CSV结构大概是这样：

• customers.csv：包含customer_id、customer_name等客户唯一信息
• address.csv：包含address_id、address_line、city等地址信息
• customer-address.csv：包含customer_id和address_id的映射关系

那正确的合并步骤应该是这样：

import pandas as pd

# 1. 读取所有CSV文件
customers = pd.read_csv('customers.csv')
addresses = pd.read_csv('address.csv')
customer_address_map = pd.read_csv('customer-address.csv')

# 2. 先合并映射表和地址表，拿到每个客户对应的所有地址
address_map_merged = pd.merge(
    customer_address_map,
    addresses,
    on='address_id',  # 按地址ID关联
    how='left'  # 保留所有映射记录，即使地址表中没有匹配（可选，根据需求调整）
)

# 3. 再和客户表合并，得到完整的客户-地址关联数据
final_result = pd.merge(
    customers,
    address_map_merged,
    on='customer_id',  # 按客户ID关联
    how='left'  # 保留所有客户，即使没有地址记录
)

这样处理后，一个客户对应多个地址的话，会生成多行记录，每个地址占一行，完全不会出现覆盖的情况。比如客户ID为1有两个地址，结果里就会有两行客户ID=1的记录，分别对应两个不同的地址。

进阶：聚合多行结果（避免重复客户行） #

如果你不想看到重复的客户行，而是想把同一个客户的所有地址合并到一个字段里（比如用逗号分隔），可以用groupby+agg来聚合：

# 按客户ID分组，聚合地址信息
aggregated_result = final_result.groupby('customer_id').agg(
    # 客户的唯一信息（比如姓名）取第一个即可，因为同一个ID的信息应该一致
    customer_name=('customer_name', 'first'),
    # 把所有地址行合并成一个字符串，去重并过滤空值
    full_addresses=('address_line', lambda x: ' | '.join(x.dropna().unique()))
).reset_index()

这样处理后，每个客户只会占一行，所有地址都合并在full_addresses字段里，方便后续查看或导出。

大型CSV的优化处理 #

如果你的CSV文件特别大，直接读入内存会溢出，那就用分块读取的方式处理：

import pandas as pd

chunk_size = 10000  # 每次读取1万行，可根据内存调整

# 先处理地址表和映射表的合并
address_chunks = []
for addr_chunk in pd.read_csv('address.csv', chunksize=chunk_size):
    merged_chunk = pd.merge(customer_address_map, addr_chunk, on='address_id', how='left')
    address_chunks.append(merged_chunk)
address_map_merged = pd.concat(address_chunks)

# 再分块处理客户表的合并
final_chunks = []
for cust_chunk in pd.read_csv('customers.csv', chunksize=chunk_size):
    merged_chunk = pd.merge(cust_chunk, address_map_merged, on='customer_id', how='left')
    final_chunks.append(merged_chunk)
final_result = pd.concat(final_chunks)

这种方式会把大文件拆成小批次处理，避免内存压力，结果和直接读取是完全一致的。

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