Express+TypeORM+MySQL下10万+行CSV快速导入5张关联表方案问询

针对你的场景，之前并行小批量插入慢的核心问题是重复数据处理冗余、数据库交互开销大、未利用MySQL批量插入特性，下面是经过验证的最优步骤：

一、前置准备 #

1. 给实体加唯一索引：避免重复数据，同时支持upsert操作

   • UserEntity：给email字段添加唯一约束（@Column({ unique: true })）
   • CompanyEntity：给name+description加联合唯一索引（@Index(['name', 'description'], { unique: true })）
   • EmployeePositionEntity：给name字段添加唯一约束（@Column({ unique: true })）
   • EmployeeEntity：主键为employee_unique_id（@PrimaryColumn()），天然唯一无需额外操作

2. 临时调整MySQL配置（批量完成后可改回默认）

   SET GLOBAL innodb_buffer_pool_size = 8*1024*1024*1024; -- 按服务器内存调整，建议占总内存50%
   SET GLOBAL innodb_log_file_size = 1*1024*1024*1024;
   SET GLOBAL innodb_flush_log_at_trx_commit = 2; -- 牺牲少量持久性换插入速度
   SET GLOBAL max_allowed_packet = 64*1024*1024; -- 避免大批次插入时触发数据包超限
   

二、核心流程：流式解析+缓存去重+批量插入 #

1. 流式解析CSV，缓存唯一数据 #

用csv-parser流式读取文件（避免10万行数据占满内存），同时用Map缓存所有唯一的基础数据，彻底避免重复插入：

import { createReadStream } from 'fs';
import csv from 'csv-parser';
import { pipeline } from 'stream/promises';

// 缓存：键为唯一标识，值为实体数据/后续要补的ID
const companyMap = new Map<string, { name: string; description: string; id?: number }>();
const positionMap = new Map<string, { name: string; id?: number }>();
const userMap = new Map<string, { name: string; email: string; id?: number }>();
const employeeMap = new Map<string, { name: string; uniqueId: string; companyKey: string; positionKey: string }>();
const reviews: Array<{ text: string; rating: number; userKey: string; employeeKey: string }> = [];

// 流式处理CSV文件
await pipeline(
  createReadStream('./reviews.csv'),
  csv(),
  async function* (source) {
    for await (const row of source) {
      // 缓存Company：用name+description作为唯一键
      const companyKey = `${row.company}-${row.company_description}`;
      if (!companyMap.has(companyKey)) {
        companyMap.set(companyKey, {
          name: row.company,
          description: row.company_description
        });
      }

      // 缓存EmployeePosition：用name作为唯一键
      const positionKey = row.employee_position;
      if (!positionMap.has(positionKey)) {
        positionMap.set(positionKey, { name: positionKey });
      }

      // 缓存User：用email作为唯一键
      const userKey = row.email;
      if (!userMap.has(userKey)) {
        userMap.set(userKey, {
          name: row.reviewer,
          email: userKey
        });
      }

      // 缓存Employee：用employee_unique_id作为唯一键
      const employeeKey = row.employee_unique_id;
      if (!employeeMap.has(employeeKey)) {
        employeeMap.set(employeeKey, {
          name: row.employee,
          uniqueId: employeeKey,
          companyKey,
          positionKey
        });
      }

      // 收集Review数据
      reviews.push({
        text: row.review,
        rating: parseInt(row.rating),
        userKey,
        employeeKey
      });
    }
  }
);

2. 事务内批量插入所有数据 #

把所有操作放在一个事务里，临时关闭外键/唯一检查，大幅降低数据库操作开销：

import { getConnection } from 'typeorm';

await getConnection().transaction(async manager => {
  // 临时关闭约束检查，插入速度提升明显
  await manager.query('SET FOREIGN_KEY_CHECKS=0');
  await manager.query('SET UNIQUE_CHECKS=0');

  // --- 第一步：插入Company并更新缓存ID ---
  const companyValues = Array.from(companyMap.values());
  await manager.createQueryBuilder()
    .insert()
    .into('company') // 直接操作表名，跳过实体实例化开销
    .values(companyValues)
    .orUpdate(['name', 'description'], ['name', 'description']) // 重复数据时更新（或用orIgnore）
    .execute();
  // 查询所有Company的ID，更新缓存
  const companies = await manager.getRepository('company').find({
    where: companyValues.map(c => ({ name: c.name, description: c.description }))
  });
  companies.forEach(c => {
    const key = `${c.name}-${c.description}`;
    companyMap.get(key).id = c.id;
  });

  // --- 第二步：插入EmployeePosition并更新缓存ID ---
  const positionValues = Array.from(positionMap.values());
  await manager.createQueryBuilder()
    .insert()
    .into('employee_position')
    .values(positionValues)
    .orUpdate(['name'], ['name'])
    .execute();
  const positions = await manager.getRepository('employee_position').find({
    where: positionValues.map(p => ({ name: p.name }))
  });
  positions.forEach(p => {
    positionMap.get(p.name).id = p.id;
  });

  // --- 第三步：插入Employee ---
  const employeeValues = Array.from(employeeMap.values()).map(e => ({
    id: e.uniqueId,
    name: e.name,
    companyId: companyMap.get(e.companyKey).id,
    positionId: positionMap.get(e.positionKey).id
  }));
  await manager.createQueryBuilder()
    .insert()
    .into('employee')
    .values(employeeValues)
    .orIgnore() // 主键重复直接忽略
    .execute();

  // --- 第四步：插入User并更新缓存ID ---
  const userValues = Array.from(userMap.values());
  await manager.createQueryBuilder()
    .insert()
    .into('user')
    .values(userValues)
    .orUpdate(['name'], ['email'])
    .execute();
  const users = await manager.getRepository('user').find({
    where: userValues.map(u => ({ email: u.email }))
  });
  users.forEach(u => {
    userMap.get(u.email).id = u.id;
  });

  // --- 第五步：批量插入Review（大批次拆分） ---
  const batchSize = 2000; // 根据MySQL配置调整，越大插入速度越快
  for (let i = 0; i < reviews.length; i += batchSize) {
    const batch = reviews.slice(i, i + batchSize);
    const reviewValues = batch.map(r => ({
      text: r.text,
      rating: r.rating,
      fromUserId: userMap.get(r.userKey).id,
      forEmployeeId: employeeMap.get(r.employeeKey).uniqueId
    }));
    await manager.createQueryBuilder()
      .insert()
      .into('review')
      .values(reviewValues)
      .execute();
  }

  // 恢复约束检查
  await manager.query('SET FOREIGN_KEY_CHECKS=1');
  await manager.query('SET UNIQUE_CHECKS=1');
});

三、关键优化点说明 #

1. 流式解析：避免一次性加载10万行数据到内存，降低内存占用
2. 缓存去重：只插入唯一的基础数据，减少数据库写入量
3. 大批次插入：用2000条/批替代100条/批，大幅减少数据库交互次数
4. 直接操作表名：跳过TypeORM实体实例化步骤，减少额外开销
5. 事务+关闭约束：事务减少提交次数，关闭约束避免插入时的冗余检查
6. Upsert操作：避免先查询再插入的冗余步骤，直接用orUpdate/orIgnore处理重复数据

这样处理下来，10万+数据的插入速度能提升5-10倍，具体取决于服务器配置。

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