如何高效将TXT文件数据合并至目标表（含Upsert+Delete）

针对全量INSERT ON CONFLICT触发百万级无效更新的问题，核心优化思路是先筛选出真正需要变更的数据集（新增+更新），仅对这部分执行Upsert；删除操作则通过主键批量匹配减少扫描开销，具体实现如下：

一、前提条件 #

确保目标表有唯一主键/唯一约束（比如id），这是数据对比的核心依据。

二、具体实现步骤 #

1. 提取数据库现有数据的主键与哈希值 #

从目标表拉取所有主键id，并计算每行数据的哈希值（用于快速判断数据是否变更）：

-- 按实际列顺序拼接，确保和后续Python计算逻辑一致
SELECT 
    id, 
    md5(concat_ws('|', col1, col2, col3, col4)) AS data_hash 
FROM target_table;

将结果读取为Pandas DataFrame（记为db_df）。

2. 处理TXT文件生成待对比数据集 #

读取TXT文件到DataFrame（记为txt_df），同步计算每行的哈希值：

import pandas as pd
import hashlib

def compute_row_hash(row, cols):
    # 按指定列顺序拼接字符串，避免因列顺序不同导致哈希偏差
    concat_str = '|'.join(str(row[col]) for col in cols)
    return hashlib.md5(concat_str.encode()).hexdigest()

# 读取TXT（根据实际分隔符调整sep参数）
txt_df = pd.read_csv('daily_data.txt', sep='\t')
# 指定需要参与哈希计算的列（和SQL中的列顺序完全一致）
target_cols = ['col1', 'col2', 'col3', 'col4']
txt_df['data_hash'] = txt_df.apply(lambda x: compute_row_hash(x, target_cols), axis=1)

3. 筛选需Upsert的增量数据 #

通过主键关联对比，筛选出新增行（数据库无对应主键）和更新行（主键存在但哈希值不同）：

# 按主键合并两个数据集，标记数据库侧的匹配情况
merged_df = pd.merge(
    txt_df, 
    db_df[['id', 'data_hash']], 
    on='id', 
    how='left', 
    suffixes=('_txt', '_db')
)

# 筛选增量数据集：新增（data_hash_db为空）或更新（哈希值不匹配）
upsert_df = merged_df[
    merged_df['data_hash_db'].isna() | 
    (merged_df['data_hash_txt'] != merged_df['data_hash_db'])
]

# 保留需要写入数据库的原始列（移除哈希辅助列）
upsert_df = upsert_df[['id'] + target_cols]

4. 执行增量Upsert #

仅对筛选出的upsert_df执行INSERT ON CONFLICT，避免全量无效更新：

from sqlalchemy import create_engine

# 初始化数据库连接（以PostgreSQL为例，其他数据库调整连接字符串）
engine = create_engine('postgresql://user:password@host:port/db_name')

# 构造Upsert SQL语句
upsert_sql = """
INSERT INTO target_table (id, col1, col2, col3, col4)
VALUES (%s, %s, %s, %s, %s)
ON CONFLICT (id) DO UPDATE SET
    col1 = EXCLUDED.col1,
    col2 = EXCLUDED.col2,
    col3 = EXCLUDED.col3,
    col4 = EXCLUDED.col4;
"""

# 分批次执行（避免单批次数据量过大导致数据库压力）
with engine.connect() as conn:
    transaction = conn.begin()
    try:
        # 按1000行一批处理，可根据数据库性能调整
        for chunk in upsert_df.to_records(index=False):
            conn.execute(upsert_sql, chunk)
        transaction.commit()
    except Exception as e:
        transaction.rollback()
        raise e

5. 高效执行Delete操作 #

删除数据库中不存在于TXT文件的行，推荐使用临时表关联的方式（避免全表扫描）：

# 将TXT中的所有主键传入数据库，创建临时表
with engine.connect() as conn:
    transaction = conn.begin()
    try:
        # 创建临时表存储有效主键
        conn.execute("CREATE TEMP TABLE temp_valid_ids (id INT PRIMARY KEY);")
        # 批量插入有效主键
        conn.execute(
            "INSERT INTO temp_valid_ids (id) SELECT unnest(%s);",
            (txt_df['id'].tolist(),)
        )
        # 删除不在临时表中的无效行
        conn.execute("DELETE FROM target_table WHERE id NOT IN (SELECT id FROM temp_valid_ids);")
        transaction.commit()
    except Exception as e:
        transaction.rollback()
        raise e

三、额外优化点 #

• 索引优化：确保主键id有主键索引，若哈希值频繁使用可添加普通索引
• 事务控制：将Upsert和Delete放在同一事务中，保证数据一致性
• 内存优化：若TXT文件过大，使用pd.read_csv的chunksize参数分批次读取处理
• 哈希计算优化：如果TXT生成时能直接附带哈希值，可省去Python侧的计算步骤

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