数据仓库（DWH）临时表与正式表主键冲突解决方案咨询

你考虑的新增时间戳字段并基于此执行upsert的方案，是数据仓库增量同步场景中的常规可行实践，但需要注意几个关键细节：

• 优先使用ERP返回的last_modified_at（而非自己生成的insert date）作为判断依据，因为ERP原生的修改时间能准确反映数据的实际更新节点，避免脚本执行延迟导致的时间偏差。
• 执行upsert时，需严格限定逻辑：仅当staging表中数据的时间戳晚于final表对应记录的时间戳时，才执行更新操作；若主键不存在则直接插入。

以下是适配Python项目的其他主流解决方案：

方案1：利用数据库原生Upsert语法 #

不同数据库都提供了原生的upsert支持，性能远优于Python层面的逻辑处理，适合数据量较大的场景：

• PostgreSQL（ON CONFLICT DO UPDATE）：

INSERT INTO final_invoices (invoice_no, amount, customer_id, last_modified_at)
SELECT invoice_no, amount, customer_id, last_modified_at FROM staging_invoices
ON CONFLICT (invoice_no) DO UPDATE SET
    amount = EXCLUDED.amount,
    customer_id = EXCLUDED.customer_id,
    last_modified_at = EXCLUDED.last_modified_at
WHERE EXCLUDED.last_modified_at > final_invoices.last_modified_at;

Python中可通过psycopg2或SQLAlchemy Core直接执行该SQL，无需额外处理数据对比逻辑。

• MySQL（ON DUPLICATE KEY UPDATE）：

INSERT INTO final_invoices (invoice_no, amount, customer_id, last_modified_at)
SELECT invoice_no, amount, customer_id, last_modified_at FROM staging_invoices
ON DUPLICATE KEY UPDATE
    amount = VALUES(amount),
    customer_id = VALUES(customer_id),
    last_modified_at = VALUES(last_modified_at)
WHERE VALUES(last_modified_at) > final_invoices.last_modified_at;

方案2：基于ERP API的精准增量拉取 #

如果ERP API支持按时间戳过滤，建议记录上次同步的最大时间戳，替代固定拉取近24小时数据的逻辑：

• 每次脚本执行前，从final表中读取当前最大的last_modified_at值，以此作为API请求的过滤条件，只拉取该时间点后修改的发票数据。
• 优势：减少无效数据拉取，避免重复处理已同步过的旧数据，尤其适合ERP中存在高频更新的场景。
• Python示例代码：

import psycopg2
import requests

# 从DWH获取上次同步的最新时间戳
conn = psycopg2.connect("dbname=dwh user=your_user")
cur = conn.cursor()
cur.execute("SELECT COALESCE(MAX(last_modified_at), '1970-01-01') FROM final_invoices;")
last_sync_time = cur.fetchone()[0]
cur.close()
conn.close()

# 调用ERP API拉取增量数据
api_params = {"modified_after": last_sync_time.isoformat()}
response = requests.get("https://your-erp-api.com/invoices", params=api_params)
incremental_invoices = response.json()

# 将数据写入staging表（省略具体存表逻辑）

方案3：Python层面的增量判断（适合小数据量） #

如果同步的数据量较小，可在Python中先完成数据对比，再执行插入/更新：

• 从final表批量读取所有发票号及其对应last_modified_at，存入字典{invoice_no: latest_modified_time}。
• 遍历staging表数据：若发票号不在字典中则执行插入；若存在且staging数据的时间戳更新，则执行更新。
• 缺点：数据量大时会占用较多内存，效率远低于数据库原生upsert。

通用注意事项 #

• 所有同步操作需包裹在数据库事务中，避免部分数据操作失败导致的一致性问题。
• 定期清理staging表的历史数据，比如删除30天前的记录，防止表膨胀。
• 若ERP的修改时间字段不可靠（如部分更新未触发时间戳变化），可改用数据哈希值：给每条发票生成MD5哈希并存入final表，同步时对比哈希值，不同则执行更新。

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