Spara en mycket stor CSV till mongoDB med mongoose

Välkommen till streaming. Vad du verkligen vill ha är en "händelseström" som bearbetar din inmatning "en bit i taget", och naturligtvis helst med en gemensam avgränsare som tecknet "nyrad" som du använder för närvarande.

För riktigt effektiva saker kan du lägga till användning av MongoDB "Bulk API" insatser för att göra din laddning så snabb som möjligt utan att äta upp allt maskinminne eller CPU-cykler.

Förespråkar inte eftersom det finns olika lösningar tillgängliga, men här är en lista som använder line- input-stream-paket för att göra "linjeavslutningsdelen" enkel.

Schemadefinitioner endast med "exempel":

var LineInputStream = require("line-input-stream"),
    fs = require("fs"),
    async = require("async"),
    mongoose = require("mongoose"),
    Schema = mongoose.Schema;

var entrySchema = new Schema({},{ strict: false })

var Entry = mongoose.model( "Schema", entrySchema );

var stream = LineInputStream(fs.createReadStream("data.txt",{ flags: "r" }));

stream.setDelimiter("\n");

mongoose.connection.on("open",function(err,conn) { 

    // lower level method, needs connection
    var bulk = Entry.collection.initializeOrderedBulkOp();
    var counter = 0;

    stream.on("error",function(err) {
        console.log(err); // or otherwise deal with it
    });

    stream.on("line",function(line) {

        async.series(
            [
                function(callback) {
                    var row = line.split(",");     // split the lines on delimiter
                    var obj = {};             
                    // other manipulation

                    bulk.insert(obj);  // Bulk is okay if you don't need schema
                                       // defaults. Or can just set them.

                    counter++;

                    if ( counter % 1000 == 0 ) {
                        stream.pause();
                        bulk.execute(function(err,result) {
                            if (err) callback(err);
                            // possibly do something with result
                            bulk = Entry.collection.initializeOrderedBulkOp();
                            stream.resume();
                            callback();
                        });
                    } else {
                        callback();
                    }
               }
           ],
           function (err) {
               // each iteration is done
           }
       );

    });

    stream.on("end",function() {

        if ( counter % 1000 != 0 )
            bulk.execute(function(err,result) {
                if (err) throw err;   // or something
                // maybe look at result
            });
    });

});

Så generellt sett bryter "stream"-gränssnittet där "nedgången" för att bearbeta "en rad i taget". Det hindrar dig från att ladda allt på en gång.

Huvuddelarna är "Bulk Operations API" från MongoDB. Detta gör att du kan "köa" många operationer åt gången innan du faktiskt skickar till servern. Så i det här fallet med användning av en "modulo" skickas skrivningar endast per 1000 bearbetade poster. Du kan verkligen göra vad som helst upp till 16 MB BSON-gränsen, men håll det hanterbart.

Förutom att operationerna bearbetas i bulk, finns det en extra "begränsare" på plats från async bibliotek. Det är egentligen inte nödvändigt, men detta säkerställer att i princip inte mer än "modulo-gränsen" av dokument är i bearbetning när som helst. De allmänna batch-"insättningarna" kommer utan någon annan IO-kostnad än minne, men "exekverings"-anropen betyder att IO bearbetas. Så vi väntar istället för att köa fler saker.

Det finns säkert bättre lösningar du kan hitta för att "strömbearbeta" data av CSV-typ, vilket det verkar vara. Men i allmänhet ger detta dig koncepten för hur du gör detta på ett minneseffektivt sätt utan att behöva äta CPU-cykler också.