如何在NestJS中使用TypeORM实现Excel导入场景下的批量更新

1. 数据库原生Upsert（优先推荐） #

绝大多数主流数据库（PostgreSQL、MySQL、SQL Server等）都支持原生Upsert语法，这比应用层循环更新效率高得多。核心逻辑是：

• 将导入的Excel数据先存入临时表（内存临时表或物理临时表均可）
• 借助数据库的Upsert语句，基于唯一标识字段（如业务ID、唯一编码）对比数据，仅更新存在差异的字段

MySQL示例 #

假设业务表为user_info，唯一键是user_id，临时表为temp_user_info，仅更新修改过的name和phone字段：

INSERT INTO user_info (user_id, name, phone, create_time)
SELECT user_id, name, phone, create_time FROM temp_user_info
ON DUPLICATE KEY UPDATE
  name = CASE WHEN temp_user_info.name != user_info.name THEN temp_user_info.name ELSE user_info.name END,
  phone = CASE WHEN temp_user_info.phone != user_info.phone THEN temp_user_info.phone ELSE user_info.phone END;

PostgreSQL示例 #

结合ON CONFLICT和EXCLUDED关键字实现差异更新：

INSERT INTO user_info (user_id, name, phone)
SELECT user_id, name, phone FROM temp_user_info
ON CONFLICT (user_id) DO UPDATE SET
  name = CASE WHEN EXCLUDED.name != user_info.name THEN EXCLUDED.name ELSE user_info.name END,
  phone = CASE WHEN EXCLUDED.phone != user_info.phone THEN EXCLUDED.phone ELSE user_info.phone END;

这种方式把数据对比和更新交给数据库处理，数据量越大，效率优势越明显。

2. 应用层预筛选差异数据 #

如果受权限或架构限制无法使用临时表，可以在应用层先筛选出与数据库有差异的行，再批量更新：

• 批量拉取数据库中对应唯一标识的现有数据（比如用SELECT * FROM 表名 WHERE 唯一键 IN (导入的唯一键列表)）
• 将导入的Excel数据与数据库数据按唯一标识映射对比，仅保留字段有变化的行
• 用ORM的批量更新方法（如SQLAlchemy的bulk_update_mappings、Django的bulk_update）一次性提交差异行

Python示例（SQLAlchemy） #

import pandas as pd
from sqlalchemy import create_engine, text

# 1. 导入Excel数据
df = pd.read_excel('updated_data.xlsx')
imported_data = df.to_dict('records')

# 2. 拉取数据库现有数据
engine = create_engine('postgresql://user:pass@host/db')
with engine.connect() as conn:
    user_ids = [item['user_id'] for item in imported_data]
    existing_data = conn.execute(text("SELECT user_id, name, phone FROM user_info WHERE user_id IN :ids"), {'ids': user_ids}).fetchall()
    existing_dict = {row[0]: {'name': row[1], 'phone': row[2]} for row in existing_data}

# 3. 筛选差异行
update_rows = []
for item in imported_data:
    existing = existing_dict.get(item['user_id'])
    if existing and (item['name'] != existing['name'] or item['phone'] != existing['phone']):
        update_rows.append(item)

# 4. 批量更新
if update_rows:
    with engine.begin() as conn:
        conn.execute(
            text("UPDATE user_info SET name = :name, phone = :phone WHERE user_id = :user_id"),
            update_rows
        )

这种方式减少了需要更新的行数，比循环单条更新效率提升显著。

3. 优化细节 #

• 导入Excel时尽量保留数据原始类型（如日期、数字），避免类型转换导致的无意义差异（比如字符串"123"和数字123被误判为不同）
• 如果数据库有last_modified字段，可以在导入时带上Excel中的修改时间，仅更新last_modified晚于数据库记录的行，进一步缩小更新范围

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