Написание больших файлов с помощью Node.JS

Идея drain заключается в том, что вы можете использовать его для тестирования здесь:

var fs = require('fs');
var stream = fs.createWriteStream('someFile.txt', {flags: 'w'});

var lines;
while (lines = getLines()) {
    for (var i = 0; i < lines.length; i++) {
        stream.write(lines[i]); //<-- the place to test
    }
}

Которым ты не являешься. Поэтому вам нужно будет реархитектировать, чтобы сделать его "реентерабельным".

var fs = require('fs');
var stream = fs.createWriteStream('someFile.txt', {flags: 'w'});

var lines;
while (lines = getLines()) {
    for (var i = 0; i < lines.length; i++) {
        var written = stream.write(lines[i]); //<-- the place to test
        if (!written){
           //do something here to wait till you can safely write again
           //this means prepare a buffer and wait till you can come back to finish
           //  lines[i] -> remainder
        }
    }
}

Однако означает ли это, что вам нужно продолжать буферизацию getLines, пока вы ждете?

var fs = require('fs');
var stream = fs.createWriteStream('someFile.txt', {flags: 'w'});

var lines,
    buffer = {
     remainingLines = []
    };
while (lines = getLines()) {
    for (var i = 0; i < lines.length; i++) {
        var written = stream.write(lines[i]); //<-- the place to test
        if (!written){
           //do something here to wait till you can safely write again
           //this means prepare a buffer and wait till you can come back to finish
           //  lines[i] -> remainder
           buffer.remainingLines = lines.slice(i);
           break;
           //notice there's no way to re-run this once we leave here.
        }
    }
}

stream.on('drain',function(){
  if (buffer.remainingLines.length){
    for (var i = 0; i < buffer.remainingLines.length; i++) {
      var written = stream.write(buffer.remainingLines[i]); //<-- the place to test
      if (!written){
       //do something here to wait till you can safely write again
       //this means prepare a buffer and wait till you can come back to finish
       //  lines[i] -> remainder
       buffer.remainingLines = lines.slice(i);
      }
    }
  }
});

Я обнаружил, что потоки являются плохим способом работы с большими файлами - это потому, что вы не можете установить адекватный размер входного буфера (по крайней мере, я не знаю хорошего способа сделать это). Вот что я делаю:

var fs = require('fs');

var i = fs.openSync('input.txt', 'r');
var o = fs.openSync('output.txt', 'w');

var buf = new Buffer(1024 * 1024), len, prev = '';

while(len = fs.readSync(i, buf, 0, buf.length)) {

    var a = (prev + buf.toString('ascii', 0, len)).split('\n');
    prev = len === buf.length ? '\n' + a.splice(a.length - 1)[0] : '';

    var out = '';
    a.forEach(function(line) {

        if(!line)
            return;

        // do something with your line here

        out += line + '\n';
    });

    var bout = new Buffer(out, 'ascii');
    fs.writeSync(o, bout, 0, bout.length);
}

fs.closeSync(o);
fs.closeSync(i);

[Edit] обновленный узел.js writable.write(...) API docs говорят:

[возвращаемое значение] является строго рекомендательным. Вы можете продолжать писать, даже если он возвращает false. Однако записи будут буферизованы в памяти, поэтому лучше не делать этого чрезмерно. Вместо этого дождитесь события слива, прежде чем записывать дополнительные данные.

[Оригинал] из stream.write(...) документация (Курсив мой):

Возвращает true, если строка была сброшен в буфер ядра. Возвращает false, чтобы указать, что буфер ядра заполнен, и данные будут отправлены в будущем.

Я интерпретирую это так, что функция "write" возвращает true, Если данная строка была немедленно записана в базовый буфер ОС, или false, если она еще не была записана, но будет записана функцией write (например, предположительно была буферизована для вас WriteStream), так что вам не нужно вызывать функцию " write" снова.

Самый простой способ справиться с этим - сделать ваш генератор строкчитаемым потоком - назовем его lineReader. Тогда следующие будут автоматически обрабатывать буферы и сливать красиво для вас:

lineReader.pipe(fs.createWriteStream('someFile.txt'));

Если вы не хотите создавать читаемый поток, вы можете прослушать вывод write для заполнения буфера и ответить следующим образом:

var i = 0, n = lines.length;
function write () {
  if (i === n) return;  // A callback could go here to know when it's done.
  while (stream.write(lines[i++]) && i < n);
  stream.once('drain', write);
}
write();  // Initial call.

Более длинный пример этой ситуации можно найтиздесь .

Несколько предложенных ответов на этот вопрос совершенно упустили из виду пункт о потоках.

Этот модуль может помочь https://www.npmjs.org/package/JSONStream

Однако предположим, что ситуация описана, и напишем код сами. Вы читаете из MongoDB как поток, с ObjectMode = true по умолчанию.

Это приведет к проблемам, если вы попытаетесь напрямую передать поток в файл-что-то вроде ошибки "недопустимый нестроковый/буферный фрагмент".

Решение этой проблемы очень простое.

Просто поместите другое преобразование между читаемым и записываемым, чтобы адаптировать объект, читаемый, к строке, записываемой соответствующим образом.

Пример Решения Кода:

var fs = require('fs'),
    writeStream = fs.createWriteStream('./out' + process.pid, {flags: 'w', encoding: 'utf-8' }),
    stream = require('stream'),
    stringifier = new stream.Transform();
stringifier._writableState.objectMode = true;
stringifier._transform = function (data, encoding, done) {
    this.push(JSON.stringify(data));
    this.push('\n');
    done();
}
rowFeedDao.getRowFeedsStream(merchantId, jobId)
.pipe(stringifier)
.pipe(writeStream).on('error', function (err) {
   // handle error condition
}

Если у вас нет входного потока, вы не можете легко использовать канал. Ничего из вышеперечисленного не сработало для меня, событие слива не срабатывает. Решается следующим образом (на основе ответа Тайлера):

var lines[]; // some very large array
var i = 0;

function write() {
    if (i < lines.length)  {
        wstream.write(lines[i]), function(err){
            if (err) {
                console.log(err);
            } else {
                i++;
                write();
            }
        });
    } else {
        wstream.end();
        console.log("done");
    }
};
write();