使用node-stream-to-mongo-db导入CSV至MongoDB时出现BSON类型错误

环境：Ubuntu系统，Node.js 18.7
需求：将一批CSV文件解析为对象后导入MongoDB，参考node-stream-to-mongo-db的示例4（本地文件流）操作，已在本地MongoDB的local数据库创建devTestOutput集合。

运行代码时出现错误：

MongoError: BSON field 'insert.documents.0' is the wrong type 'binData', expected type 'object'

我的代码如下：

const csv = require('csvtojson');
const streamToMongoDB = require('stream-to-mongo-db').streamToMongoDB;

// where the data will end up
// const outputDBConfig = { dbURL : 'mongodb://localhost:27017/streamToMongoDB', collection : 'devTestOutput' };
const outputDBConfig = { dbURL: 'mongodb://localhost:27017/local', collection: 'devTestOutput' };

// create the writable stream
const writableStream = streamToMongoDB(outputDBConfig);


const main = async () => {
    const readFileStream = fs.createReadStream(path);
    const writeFileStream = fs.createWriteStream(__dirname + '/file2');

    await readFileStream.pipe(csv(), { objectMode: true }).on('data', x => console.log(x.toString())).on('error', err => console.log(err))
        .pipe(writableStream);
    console.log('finish');


}

main()

请问我哪里操作出错了？

报错核心原因：传递给MongoDB写入流的不是JSON对象，而是二进制/字符串数据，导致MongoDB无法识别为合法文档对象。具体问题及修正如下：

1. 关键错误点 #

• pipe参数位置错误：{ objectMode: true }应该作为csv()的配置项，而非pipe方法的参数；
• 无效的await使用：pipe返回流对象而非Promise，await无法正确等待流处理完成；
• 缺失必要模块：代码未引入fs和path模块，会导致文件读取报错；
• 无意义的数据流转换：x.toString()将解析后的对象转为字符串输出，虽不直接修改流数据，但易混淆数据类型，且冗余。

2. 修正后的代码 #

const fs = require('fs');
const path = require('path');
const csv = require('csvtojson');
const streamToMongoDB = require('stream-to-mongo-db').streamToMongoDB;

// MongoDB配置
const outputDBConfig = { 
  dbURL: 'mongodb://localhost:27017/local', 
  collection: 'devTestOutput' 
};

// 创建MongoDB写入流
const writableStream = streamToMongoDB(outputDBConfig);

const main = () => {
  // 替换为你的目标CSV文件路径
  const csvFilePath = path.join(__dirname, 'your-input.csv');
  const readFileStream = fs.createReadStream(csvFilePath);

  // 正确的流处理流程
  readFileStream
    .pipe(csv({ objectMode: true })) // 确保CSV转为对象流输出
    .on('data', (obj) => console.log('解析后的文档:', obj))
    .on('error', (err) => console.error('CSV解析错误:', err))
    .pipe(writableStream)
    .on('finish', () => console.log('数据导入完成'))
    .on('error', (err) => console.error('MongoDB写入错误:', err));
};

main();

3. 额外注意事项 #

• 确认CSV文件格式合法，无语法错误；
• 确保MongoDB服务正常运行，local数据库及devTestOutput集合已存在；
• 批量处理多CSV文件时，可循环遍历文件路径，复用上述流处理逻辑。

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