在Azure Synapse Analytics或Azure Data Factory中按ID对两个CSV执行Upsert

核心思路 #

1. 分别清洗两个CSV：对每个文件，按ID分组，保留最后修改时间最新的行（解决同ID多行的重复数据问题）
2. 合并处理后的数据：
   • 优先保留csv2中的数据（实现ID匹配时的更新逻辑）
   • 补充csv1中ID未出现在csv2里的数据（实现新ID的插入逻辑）

代码实现 #

假设你的CSV字段结构如下（可根据实际字段名调整）：

• csv1字段：IDcsv1, 字段A, 字段B, 最后修改时间
• csv2字段：IDcsv2, 字段A, 字段B, 最后修改时间

import pandas as pd

# 1. 读取CSV文件，替换为你的实际文件路径
csv1_path = "csv1.csv"
csv2_path = "csv2.csv"
df1 = pd.read_csv(csv1_path)
df2 = pd.read_csv(csv2_path)

# 2. 统一ID列名，避免合并时的歧义
df1 = df1.rename(columns={"IDcsv1": "ID"})
df2 = df2.rename(columns={"IDcsv2": "ID"})

# 3. 转换时间列为datetime类型，确保能正确排序比较
# 替换"最后修改时间"为你实际的时间列名称
df1["最后修改时间"] = pd.to_datetime(df1["最后修改时间"], errors="coerce")
df2["最后修改时间"] = pd.to_datetime(df2["最后修改时间"], errors="coerce")

# 4. 每个文件内按ID分组，保留最后修改时间最新的行
# idxmax()返回每个ID组内时间最大的行的索引，再用loc提取对应行
df1_latest = df1.loc[df1.groupby("ID")["最后修改时间"].idxmax()]
df2_latest = df2.loc[df2.groupby("ID")["最后修改时间"].idxmax()]

# 5. 合并数据：优先用csv2的最新数据，补充csv1中独有的ID数据
# ~符号表示取反，筛选出csv1中ID不在csv2里的行
merged_df = pd.concat([df2_latest, df1_latest[~df1_latest["ID"].isin(df2_latest["ID"])]])

# 6. 保存结果到新CSV文件
merged_df.to_csv("合并更新后的CSV.csv", index=False, encoding="utf-8-sig")

关键细节说明 #

• ID列统一：将两个文件的ID列重命名为相同名称，是后续分组、合并的基础
• 时间列处理：errors="coerce"会把无效时间格式转为NaT，若需要保留这类行，可去掉该参数或单独处理
• 取最新行逻辑：通过分组取时间最大值对应的索引，确保每个ID只保留最新的一条数据
• 合并优先级：先放入csv2的数据，再追加csv1独有的数据，保证同ID数据被csv2覆盖（即更新）

替代方案（命令行工具） #

如果不想写代码，可使用csvkit工具（需先安装：pip install csvkit），结合shell命令实现：

# 处理csv1，按ID分组取最新行
csvsql --query "SELECT * FROM csv1 GROUP BY IDcsv1 HAVING 最后修改时间 = MAX(最后修改时间)" csv1.csv > csv1_latest.csv

# 处理csv2，按ID分组取最新行
csvsql --query "SELECT * FROM csv2 GROUP BY IDcsv2 HAVING 最后修改时间 = MAX(最后修改时间)" csv2.csv > csv2_latest.csv

# 合并：用csv2数据覆盖csv1同ID数据，补充csv1独有的数据
csvjoin --left -c IDcsv1,IDcsv2 csv1_latest.csv csv2_latest.csv | csvcut -c IDcsv2,字段A,字段B,最后修改时间 | sed 's/IDcsv2/ID/' > merged.csv

注：命令行方案要求字段名统一，且时间列格式能被SQL正确识别，灵活性不如Python方案

内容的提问来源于stack exchange，提问作者Zacarías