使用Mongoose实现MongoDB批量插入：CSV转JSON上传性能优化问询

嘿，我来帮你搞定这个CSV转MongoDB的性能问题——你现在遇到的内存飙到1.3GB、处理慢的问题，大概率是因为脚本把整个CSV文件都塞进内存里处理了，咱们从几个核心方向来优化：

1. 用流式解析代替一次性加载全量数据 #

这是解决内存占用过高的核心！不要把整个8.5MB的CSV（8.5万行）一次性读进内存解析，而是用流式解析，读一行处理一行，内存里永远只保留少量数据。

Node.js生态里推荐用csv-parser这个轻量高效的库，它支持流式处理，配合fs.createReadStream就能实现边读边解析。

2. 批量插入MongoDB，减少数据库请求开销 #

如果之前是每解析一行就调用一次insertOne，那数据库请求次数会爆炸（4个文件就是34万次请求），这绝对是速度慢的元凶之一。

改成批量插入：攒够N条数据（比如1000条）就调用一次insertMany，这样能把请求次数降到几百次，大幅减少网络IO和数据库的处理开销。注意MongoDB单批插入的文档总大小不能超过16MB，所以根据你的数据结构调整批次大小（1000条一般没问题）。

3. 简化文件操作，减少不必要的IO #

你现在用了fs.rename重命名上传的临时文件，其实完全可以跳过这一步——直接用上传后临时文件的路径（req.file.path）来创建读取流，省掉一次文件系统的重命名操作，节省时间。

4. 优化CSV解析的细节 #

• 指定明确的解析规则：比如分隔符、引号类型（比如csv({ separator: ',', quote: '"' })），避免解析器做不必要的猜测，提升速度。
• 过滤冗余字段：如果CSV里有不需要存入MongoDB的列，在解析时直接过滤掉，减少后续处理的数据量。
• 选择高效的解析库：csv-parser比一些重型库快很多，优先用它。

5. 内存与GC的小技巧 #

• 处理完每一批数据后，及时清空批次数组（batch = []），让V8的垃圾回收能及时回收内存。
• 如果你的Node.js版本支持，可以启动时加--expose-gc参数，在批次插入后手动触发GC（global.gc()），但不要频繁调用，避免影响性能。
• 避免在data事件回调里创建大量嵌套对象，尽量保持每一行数据的结构简洁。

整合后的代码示例 #

const csv = require('csv-parser');
const fs = require('fs');
const { MongoClient } = require('mongodb');
const upload = require('multer')({ dest: UPLOAD_PATH }); // 假设multer配置在这里

router.route('/upload').post(upload.single('data'), async (req, res) => {
  const BATCH_SIZE = 1000;
  let batch = [];
  let client;

  try {
    // 连接MongoDB
    client = await MongoClient.connect('你的MongoDB连接字符串');
    const db = client.db('目标数据库名');
    const collection = db.collection('目标集合名');

    // 直接读取上传的临时文件，流式解析CSV
    fs.createReadStream(req.file.path)
      .pipe(csv({
        separator: ',', // 根据你的CSV实际分隔符调整
        quote: '"'
      }))
      .on('data', (row) => {
        // 可选：过滤不需要的字段，或者转换数据类型
        const processedRow = {
          // 比如只保留需要的字段：name: row.name, age: Number(row.age)
          ...row
        };

        batch.push(processedRow);

        // 达到批次大小就批量插入
        if (batch.length >= BATCH_SIZE) {
          // 这里用异步插入，注意控制并发（可以用队列限制并发数）
          collection.insertMany(batch)
            .catch(err => console.error('批量插入失败:', err));
          batch = []; // 清空批次
        }
      })
      .on('end', async () => {
        // 插入剩余的不足一批的数据
        if (batch.length > 0) {
          await collection.insertMany(batch);
        }
        // 清理临时文件
        fs.unlinkSync(req.file.path);
        await client.close();
        res.status(200).json({ message: '所有文件上传解析完成' });
      })
      .on('error', async (err) => {
        console.error('CSV解析出错:', err);
        fs.unlinkSync(req.file.path);
        await client.close();
        res.status(500).json({ error: '解析上传文件失败' });
      });
  } catch (err) {
    console.error('初始化出错:', err);
    if (client) await client.close();
    res.status(500).json({ error: '服务端错误' });
  }
});

额外注意事项 #

• 错误处理：确保在任何出错场景下都关闭数据库连接、删除临时文件，避免资源泄漏。
• 并发控制：如果批次插入的并发过高，可能会给MongoDB带来压力，可以用一个简单的队列来限制同时进行的插入操作（比如最多2个并发）。
• 数据去重：如果CSV里可能有重复数据，可以考虑用updateMany配合upsert: true，或者在批量插入前做简单的去重（但会增加一点开销，按需选择）。

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