200万件のCSVをparseしつつMongoDBに素早く突っ込むにはBulk()を使う（Nodeでやってみた）

はじめに

• 大量のデータを操作するときは Bulk() を使おうという当たり前の話
• 言語は別になんでもいいけどなんとなくNode
• 英辞郎のCSVデータ（約 200万件 ）をmongoに突っ込んだ時のコードとか。

# コマンド例
node good.js eijiro_dic_utf8.csv

Bulk を使わない場合

bad.js

var fs = require('fs');
var MongoClient = require('mongodb').MongoClient;
var CSV = require('comma-separated-values');

var NAME_MONGO_DB = 'eijiro';
var NAME_MONGO_COLLECTION = 'words';
var url = 'mongodb://localhost:27017/' + NAME_MONGO_DB;

var inputCsv = process.argv[2];
var text = fs.readFileSync(inputCsv, 'utf-8');

// data: example [ {a : 1}, {a : 2}, {a : 3} ]
var insertDocuments = function(db, data, callback) {
    // Get the documents collection
    var collection = db.collection(NAME_MONGO_COLLECTION);
    // Insert some documents
    collection.insert(data, function(err, result) {
        if (err) return console.error(err);
        callback(result);
    });
};

MongoClient.connect(url, function(err, db) {
    if (err) return console.error(err);

    var rows = new CSV(text, { header: true, cast: false }).parse();

    insertDocuments(db, rows, function(result) {
        db.close();
    });
});

• 大体40分くらいかかった気がする。

Bulk を使う場合

good.js

var fs = require('fs');
var MongoClient = require('mongodb').MongoClient;
var CSV = require('comma-separated-values');

var NAME_MONGO_DB = 'eijiro';
var NAME_MONGO_COLLECTION = 'words';
var url = 'mongodb://localhost:27017/' + NAME_MONGO_DB;

var inputCsv = process.argv[2];
var text = fs.readFileSync(inputCsv, 'utf-8');

MongoClient.connect(url, function(err, db) {
    if (err) return console.error(err);

    var col = db.collection(NAME_MONGO_COLLECTION);
    var batch = col.initializeUnorderedBulkOp();

    new CSV(text, { header: true, cast: false }).forEach(function(obj) {
        batch.insert(obj);
    });

    batch.execute(function(err, result) {
        db.close();
        if (err) console.err(err);
        process.exit();
    });
});

• 1分もあれば終わる。

Bulkに関する補足

Each group of operations can have at most 1000 operations. If a group exceeds this limit, MongoDB will divide the group into smaller groups of 1000 or less. For example, if the bulk operations list consists of 2000 insert operations, MongoDB creates 2 groups, each with 1000 operations.

雑感

• readFileSync() を使わなければならないのが残念すぎる。
• 全部streamっぽく処理できれば良いなぁと思う。（ comma-separated-values の同期的な書き方が気持ち悪すぎる）
• 最初は、 C2FO/fast-csv を使おうとしたが、parseの際に日本語文字を上手く解釈できずに落ちたので泣く泣く止めた。
• github repository