本文介绍如何在 5 秒内将 10 万+ 行 CSV 数据高速导入 MySQL 的 5 张关联表，核心是绕过 ORM 低效逐行插入，采用原生 LOAD DATA INFILE + 批量 SELECT 插入的 staging table 方案。
本文介绍如何在 5 秒内将 10 万+ 行 csv 数据高速导入 mysql 的 5 张关联表，核心是绕过 orm 低效逐行插入，采用原生 `load data infile` + 批量 select 插入的 staging table 方案。
在 Express + TypeORM 技术栈中处理 10 万+ 行 CSV 导入时，直接使用 repository.save() 分批（如每 100 条）插入不仅耗时长，还会因外键约束、事务开销、N+1 查询及 TypeORM 的实体映射机制显著拖慢性能——实测往往需数分钟。真正满足「5 秒内完成」的关键，在于跳出应用层思维，让数据落地由 MySQL 原生能力驱动。
✅ 推荐方案：Staging Table + 原生 SQL 批量归一化
该方案分三阶段，全程在数据库内高效完成，规避 ORM 开销：
1. 创建临时原始表（staging table）
CREATE TABLE raw_import ( id INT AUTO_INCREMENT PRIMARY KEY, reviewer VARCHAR(255), review LONGTEXT, email VARCHAR(255), rating TINYINT, employee VARCHAR(255), employee_position VARCHAR(255), employee_unique_id VARCHAR(255), company VARCHAR(255), company_description LONGTEXT );
2. 使用 LOAD DATA INFILE 高速载入 CSV（毫秒级）
确保 MySQL 已启用 local_infile，并执行：
SET GLOBAL local_infile = ON; LOAD DATA LOCAL INFILE '/path/to/data.csv' INTO TABLE raw_import FIELDS TERMINATED BY ',' ENCLOSED BY '"' LINES TERMINATED BY '\n' IGNORE 1 ROWS;
✅ 优势：MySQL 原生 C 实现，无网络序列化、无 JS 对象构建，10 万行通常
3. 并行构建目标表（含自动 ID 关联）
利用 INSERT ... SELECT 一次性生成规范数据，并通过 INSERT IGNORE / ON DUPLICATE KEY UPDATE 处理重复与外键映射：
-- 插入 users（去重 + 生成 UUID 或自增 ID） INSERT INTO user (name, email) SELECT DISTINCT reviewer, email FROM raw_import WHERE email IS NOT NULL AND email != ''; -- 插入 employee_positions（去重） INSERT INTO employee_position (name) SELECT DISTINCT employee_position FROM raw_import WHERE employee_position IS NOT NULL; -- 插入 companies（去重） INSERT INTO company (name, description) SELECT DISTINCT company, company_description FROM raw_import WHERE company IS NOT NULL; -- 插入 employees（关联 position & company） INSERT INTO employee (id, name, position_id, company_id) SELECT ri.employee_unique_id, ri.employee, ep.id, c.id FROM raw_import ri JOIN employee_position ep ON ep.name = ri.employee_position JOIN company c ON c.name = ri.company; -- 插入 reviews（关联 user & employee） INSERT INTO review (text, rating, from_user_id, for_employee_id) SELECT ri.review, ri.rating, u.id, e.id FROM raw_import ri JOIN user u ON u.email = ri.email JOIN employee e ON e.id = ri.employee_unique_id;
⚠️ 注意事项：
- 所有目标表主键/唯一索引需提前建好（如 employee.id 设为 PRIMARY KEY，user.email 加 UNIQUE INDEX），否则关联查询会极慢；
- 若 user.id 是 UUID 类型，建议改用 BIGINT AUTO_INCREMENT 或 CHAR(36) + 应用层预生成，避免 MySQL 无法高效索引 UUID；
- 生产环境务必在事务中执行上述 INSERTs，或使用 RENAME TABLE 原子切换（见下文“零停机替换”）；
- TypeORM 中禁用 synchronize: true，所有 DDL 由 DBA 或迁移脚本统一管理。
? 进阶优化：零停机数据替换（
若需全量覆盖旧数据（如每日报表更新），推荐以下原子操作：
-- 1. 创建新表（带 _new 后缀） CREATE TABLE user_new LIKE user; CREATE TABLE review_new LIKE review; -- ... 其余表同理 -- 2. 将清洗后数据导入 _new 表（同上 INSERT SELECT） -- 3. 原子重命名（瞬间生效，无锁表风险） RENAME TABLE user TO user_old, user_new TO user, review TO review_old, review_new TO review, employee TO employee_old, employee_new TO employee, employee_position TO employee_position_old, employee_position_new TO employee_position, company TO company_old, company_new TO company; -- 4. 异步清理旧表 DROP TABLE user_old, review_old, ...;
❌ 为什么不推荐 TypeORM 批处理？
- save([...], { chunk: 100 }) 仍会为每批次开启事务、校验关系、生成 N 条 INSERT 语句；
- queryRunner.manager.insert() 虽稍快，但缺乏批量值绑定优化，且无法跨表原子关联；
- 外键约束强制实时校验，极大降低并发插入吞吐。
✅ 总结
| 方法 | 耗时估算（10 万行） | 是否满足 5 秒 | 维护性 | 推荐场景 |
|---|---|---|---|---|
| TypeORM 分批 save | 90–300 秒 | ❌ | 高 | 小数据量、开发调试 |
| LOAD DATA INFILE + INSERT SELECT | 1–3 秒 | ✅ | 中（需 SQL 能力） | 生产高频大批量导入 |
| RENAME TABLE 切换 | ✅ | 低（需运维协同） | 全量更新、强一致性要求 |
最终，将 CSV 解析、字段映射、冲突处理等逻辑下沉至 SQL 层，不仅能达成 5 秒目标，更能获得可复现、可审计、高性能的生产级数据管道。TypeORM 应专注业务查询与领域建模，而非承担 ETL 职责。
