MongoDB批量新增CSV数据：避免覆盖及updated_at误更新

我有一个包含170万条记录的.csv文件，其中部分记录已存在于我的MongoDB集合（该集合已有220万条记录），其余为新记录。我希望仅添加集合中不存在的记录，且不覆盖或更新现有记录，每条记录以enterprise_number作为唯一标识。

目前找到的方案都是先创建操作列表，再使用model.bulkWrite()方法执行，但这类方案使用updateOne()会更新现有记录。因此我需要一种可批量新增新记录且不更新/覆盖现有记录的方法。

初始代码 #

// Read data
const data = await readCSV('data/companies.csv');

// Format data
const formatted_data = data.map(record => ({
    name: record.name,
    enterprise_number: record.enterprise_number,
    vat_number: record.vat_number,
    legal_form: record.juridical_form || null,
    activity_codes: (!record.nace_codes || record.nace_codes === '') ? [] : JSON.parse(record.nace_codes.replace(/'/g, '"')),
    address: {
        street: record.street,
        house_number: record.house_number,
        additional: record.additional,
        postal_code: record.postal_code,
        city: record.city,
        country: record.country
    },
    phone: record.tel || null,
    mobile: record.mobile || null,
    email: record.email || null,
    establishment_date: record.start_date,
    status: record.is_active === '1' ? 'active' : 'inactive',
}));

const batchSize = 10000;
    
// Create batches of 10.000 records
for (let i = 0; i < formatted_data.length; i += batchSize) {
    let batch = formatted_data.slice(i, i + batchSize);

    let operations = batch.map((row) => ({
        updateOne: {
            filter: { enterprise_number: row.enterprise_number },
            update: row,
            upsert: true
        }
    }));

    try {
        let result = await Company.bulkWrite(operations);
        
        console.log(`Batch ${i} done: ${result.modifiedCount} records modified, ${result.upsertedCount} records upserted`);
    } catch (error) {
        console.error(`Error updating batch ${i}: ${error}`);
    }
}

console.log('All done');

改进后的代码 #

基于相关方案调整后，我将代码更新为如下版本：

for (let i = 0; i < formatted_data.length; i += batch_size) {
    let batch = formatted_data.slice(i, i + batch_size);

    let operations = batch.map((row) => ({
        updateOne: {
            filter: { enterprise_number: row.enterprise_number },
            update: { $setOnInsert: row },
            upsert: true
        }
    }));

    try {
        let result = await Kbo.bulkWrite(operations);
        console.log(`Batch ${i} done: ${result.result.nUpserted} records inserted`);
    } catch (error) {
        console.error(`Error inserting batch ${i}: ${error}`);
    }
}

该方案可新增记录且不更新现有记录的属性，但会更新所有匹配记录的updated_at字段，我需要解决此问题。

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