Pandas处理百万级数据关联与清洗

前言

在日常的数据处理工作中，我们经常面临这样的场景：手里有一份几万行的业务 Excel 表，需要去关联另外两份几十万甚至上百万行的 CSV 数据源（比如物料主数据、历史订单表），取出对应的字段追加到原表中。

如果直接用 Excel 的 VLOOKUP 或 XLOOKUP，面对百万级数据，卡顿、崩溃几乎是常态。而使用 Python 的 pandas 库，不仅速度快，逻辑也更清晰。

本文将以一个实际的业务需求为例，记录如何使用 Pandas 高效完成多表关联（Merge）、时间戳转换以及解决最常见的“数据行数膨胀”问题。

业务场景

我们需要处理三个文件：

1. 主表 (Excel)：约 2 万行，包含导入报错的订单信息。
2. 物料映射表 (CSV)：约 34 万行，用于将 ITEM_CODE 转换为 OLD_PARTNUMBER。
3. 旧订单表 (CSV)：约 133 万行，用于根据 订单号 + 旧物料号 获取历史订单的日期和行号信息。

目标：将文件 2 和 3 的信息匹配到文件 1 中，并导出结果。

核心技术点

在使用 Pandas 处理此类任务时，有几个关键点决定了脚本的健壮性和效率。

1. 内存优化与类型安全

读取百万行 CSV 时，如果全部读取，内存消耗巨大。同时，Excel 中的订单号可能是数字格式（如 123），而 CSV 读取出来可能是字符串（"123"）或者浮点数（123.0），直接关联会导致匹配失败。

最佳实践：

• 使用 usecols 只读取需要的列。
• 使用 dtype=str 强制指定关联键为字符串，防止精度丢失或类型不匹配。

# 读取百万级 CSV 的正确姿势
df_orders = pd.read_csv(
    'XX_ONT_SCHEDULING_T.csv',
    usecols=['ORDER_NUMBER', 'OLD_PARTNUMBER', 'PROMISE_DATE'], # 只读需要的列
    dtype=str, # 强制全为字符串，避免 001 变成 1，或长数字变成科学计数法
    skipinitialspace=True
)

# 清洗空格（非常重要，CSV常带有隐形空格）
df_orders['ORDER_NUMBER'] = df_orders['ORDER_NUMBER'].str.strip()

2. 处理时间戳（向量化操作）

原始数据中包含毫秒级时间戳（如 1.70969E+12），需要转为日期格式。尽量避免使用 for 循环，利用 Pandas 的向量化操作效率更高。

# 将毫秒时间戳列批量转换为日期字符串
def convert_ms_to_date(ts):
    try:
        if pd.isna(ts) or ts == '': return None
        return pd.to_datetime(ts, unit='ms').strftime('%Y-%m-%d %H:%M:%S')
    except: return None

# apply 应用到整列
df_main['日期'] = df_main['时间戳列'].apply(convert_ms_to_date)

3. 多键关联 (Merge/Join)

Pandas 的 merge 相当于 SQL 中的 JOIN。本例中需要基于两个字段确定唯一性。

# 相当于 SQL: LEFT JOIN ON t1.order_num = t2.order_num AND t1.part = t2.part
df_final = pd.merge(
    df_main,
    df_orders,
    on=['ORDER_NUMBER', 'OLD_PARTNUMBER'], # 双键关联
    how='left'
)

遇到的坑：关联后行数膨胀

这是新手最容易遇到的问题。

• 现象：主表原有 1.7 万行，关联后变成了 2.5 万行。
• 原因：这是一对多（1-to-N）关联导致的。参照表（CSV）中，关联键不唯一。例如同一个订单号和物料号，在历史表中可能有两条记录（比如曾经做过拆行处理）。
• 后果：主表的一行被复制了多次，导致数据重复统计。

解决方案：在 Merge 之前，必须对右表（参照表）进行去重。

# 核心代码：去重
# subset指定根据哪些列去重，keep='first'表示保留第一条
df_orders.drop_duplicates(
    subset=['ORDER_NUMBER', 'OLD_PARTNUMBER'], 
    keep='first', 
    inplace=True
)

# 再次关联，行数即可保持完全一致

完整代码示例

基于 Python 3.10 的完整处理逻辑如下：

import pandas as pd
import os

def process_data():
    # 1. 读取主表
    df_main = pd.read_excel('Input_File.xlsx')
    original_count = len(df_main)
    
    # 统一关联键格式
    df_main['ITEM_CODE'] = df_main['ITEM_CODE'].astype(str).str.strip()
    
    # 2. 读取参照表并去重 (关键步骤)
    df_items = pd.read_csv('Items.csv', dtype=str, usecols=['ITEM_CODE', 'OLD_PART'])
    # 确保 ITEM_CODE 唯一，否则主表会膨胀
    df_items.drop_duplicates(subset=['ITEM_CODE'], keep='first', inplace=True)
    
    # 3. 第一次关联
    df_step1 = pd.merge(df_main, df_items, on='ITEM_CODE', how='left')
    
    # 4. 读取历史订单表并去重
    df_history = pd.read_csv('History.csv', dtype=str)
    # 确保 订单号+物料号 唯一
    df_history.drop_duplicates(subset=['ORDER_NUMBER', 'OLD_PART'], keep='first', inplace=True)
    
    # 5. 第二次关联
    df_final = pd.merge(df_step1, df_history, on=['ORDER_NUMBER', 'OLD_PART'], how='left')
    
    # 6. 校验与输出
    if len(df_final) == original_count:
        print("校验通过：行数未发生变化")
        df_final.to_excel('Output_Result.xlsx', index=False)
    else:
        print(f"警告：行数发生改变 (原: {original_count} -> 现: {len(df_final)})")

if __name__ == '__main__':
    process_data()

总结

Pandas 在处理中等规模数据（百万行级别）时，是 Excel 最好的替代品。在写脚本时，只要注意以下三点，就能避开 90% 的坑：

1. 类型对齐：关联键统一转为 str 再 Merge。
2. 去重处理：在 left join 之前，务必确保右表的关联键是唯一的（drop_duplicates）。
3. 按需读取：大文件只读需要的列，节省内存。

希望这篇文章能对大家处理数据有所帮助。