Разбор огромных лог-файлов в Node.js — чтение построчно

Я искал решение для построчного анализа очень больших файлов (gbs) с использованием потока. Все сторонние библиотеки и примеры мне не подходили, так как обрабатывали файлы не построчно (типа 1, 2, 3, 4..) или читали весь файл в память

Следующее решение может анализировать очень большие файлы построчно, используя поток и канал. Для тестирования я использовал файл размером 2,1 ГБ с 17 000 000 записей. Использование оперативной памяти не превышало 60 мб.

Сначала установите пакет event-stream:

npm install event-stream

Потом:

var fs = require('fs')
    , es = require('event-stream');

var lineNr = 0;

var s = fs.createReadStream('very-large-file.csv')
    .pipe(es.split())
    .pipe(es.mapSync(function(line){

        // pause the readstream
        s.pause();

        lineNr += 1;

        // process line here and call s.resume() when rdy
        // function below was for logging memory usage
        logMemoryUsage(lineNr);

        // resume the readstream, possibly from a callback
        s.resume();
    })
    .on('error', function(err){
        console.log('Error while reading file.', err);
    })
    .on('end', function(){
        console.log('Read entire file.')
    })
);

Пожалуйста, дайте мне знать, как это происходит!

Вы можете использовать встроенный пакет readline, см. документацию здесь. Я использую stream для создания нового выходного потока.

var fs = require('fs'),
    readline = require('readline'),
    stream = require('stream');

var instream = fs.createReadStream('/path/to/file');
var outstream = new stream;
outstream.readable = true;
outstream.writable = true;

var rl = readline.createInterface({
    input: instream,
    output: outstream,
    terminal: false
});

rl.on('line', function(line) {
    console.log(line);
    //Do your stuff ...
    //Then write to outstream
    rl.write(cubestuff);
});

Обработка больших файлов займет некоторое время. Скажите, если это работает.

Мне очень понравился ответ @gerard, который на самом деле заслуживает быть здесь правильным ответом. Я сделал некоторые улучшения:

• Код находится в классе (модульном)
• Парсинг включен
• Возможность возобновления предоставляется извне в случае, если асинхронное задание связано с чтением CSV, например, вставка в БД или HTTP-запрос.
• Чтение блоков/размеров пакетов, которые пользователь может объявить. Я также позаботился о кодировании в потоке, если у вас есть файлы в другой кодировке.

Вот код:

'use strict'

const fs = require('fs'),
    util = require('util'),
    stream = require('stream'),
    es = require('event-stream'),
    parse = require("csv-parse"),
    iconv = require('iconv-lite');

class CSVReader {
  constructor(filename, batchSize, columns) {
    this.reader = fs.createReadStream(filename).pipe(iconv.decodeStream('utf8'))
    this.batchSize = batchSize || 1000
    this.lineNumber = 0
    this.data = []
    this.parseOptions = {delimiter: '\t', columns: true, escape: '/', relax: true}
  }

  read(callback) {
    this.reader
      .pipe(es.split())
      .pipe(es.mapSync(line => {
        ++this.lineNumber

        parse(line, this.parseOptions, (err, d) => {
          this.data.push(d[0])
        })

        if (this.lineNumber % this.batchSize === 0) {
          callback(this.data)
        }
      })
      .on('error', function(){
          console.log('Error while reading file.')
      })
      .on('end', function(){
          console.log('Read entirefile.')
      }))
  }

  continue () {
    this.data = []
    this.reader.resume()
  }
}

module.exports = CSVReader

Итак, в основном, вот как вы будете его использовать:

let reader = CSVReader('path_to_file.csv')
reader.read(() => reader.continue())

Я протестировал это с файлом CSV размером 35 ГБ, и это сработало для меня, и поэтому я решил построить его на @gerard. ответ, отзывы приветствуются.

Я использовал https://www.npmjs.com/package/line-by-line за чтение более 1 000 000 строк из текстового файла. При этом занимаемый объем оперативной памяти составлял около 50-60 мегабайт.

    const LineByLineReader = require('line-by-line'),
    lr = new LineByLineReader('big_file.txt');

    lr.on('error', function (err) {
         // 'err' contains error object
    });

    lr.on('line', function (line) {
        // pause emitting of lines...
        lr.pause();

        // ...do your asynchronous line processing..
        setTimeout(function () {
            // ...and continue emitting lines.
            lr.resume();
        }, 100);
    });

    lr.on('end', function () {
         // All lines are read, file is closed now.
    });

Документация Node.js предлагает очень элегантный пример использования модуля Readline.

Пример: Построчное чтение файлового потока

const fs = require('fs');
const readline = require('readline');

const rl = readline.createInterface({
    input: fs.createReadStream('sample.txt'),
    crlfDelay: Infinity
});

rl.on('line', (line) => {
    console.log(`Line from file: ${line}`);
});

Примечание: мы используем параметр crlfDelay, чтобы распознавать все экземпляры CR LF ('\r\n') как один разрыв строки.

Помимо чтения большого файла построчно, вы также можете читать его по частям. Дополнительную информацию см. в этой статье.

var offset = 0;
var chunkSize = 2048;
var chunkBuffer = new Buffer(chunkSize);
var fp = fs.openSync('filepath', 'r');
var bytesRead = 0;
while(bytesRead = fs.readSync(fp, chunkBuffer, 0, chunkSize, offset)) {
    offset += bytesRead;
    var str = chunkBuffer.slice(0, bytesRead).toString();
    var arr = str.split('\n');

    if(bytesRead = chunkSize) {
        // the last item of the arr may be not a full line, leave it to the next chunk
        offset -= arr.pop().length;
    }
    lines.push(arr);
}
console.log(lines);

У меня была такая же проблема. После сравнения нескольких модулей, вроде бы имеющих такую ​​возможность, решил сделать сам, это проще, чем я думал.

суть: https://gist.github.com/deemstone/8279565

var fetchBlock = lineByline(filepath, onEnd);
fetchBlock(function(lines, start){ ... });  //lines{array} start{int} lines[0] No.

Он покрывает файл, открытый в закрытии, возвращаемый fetchBlock() будет извлекать блок из файла, заканчиваться разделением на массив (будет обрабатывать сегмент из последней выборки).

Я установил размер блока 1024 для каждой операции чтения. Здесь могут быть ошибки, но логика кода очевидна, попробуйте сами.

node-byline использует потоки, поэтому я бы предпочел его для ваших огромных файлов.

для ваших преобразований даты я бы использовал moment.js.

для максимизации пропускной способности вы можете подумать об использовании программного кластера. есть несколько хороших модулей, которые довольно хорошо оборачивают собственный кластерный модуль узла. мне нравится cluster-master от isaacs. например вы можете создать кластер из x рабочих, которые все вычисляют файл.

для сравнения разделения и регулярных выражений используйте benchmark.js. я не тестировал его до сих пор. Benchmark.js доступен в виде узла-модуля.

Основываясь на этом ответе на вопросы, я реализовал класс, который вы можете использовать читать файл синхронно построчно с fs.readSync(). Вы можете сделать эту «паузу» и «возобновить», используя обещание Q (jQuery, похоже, требует DOM, поэтому не могу запустить его с nodejs):

var fs = require('fs');
var Q = require('q');

var lr = new LineReader(filenameToLoad);
lr.open();

var promise;
workOnLine = function () {
    var line = lr.readNextLine();
    promise = complexLineTransformation(line).then(
        function() {console.log('ok');workOnLine();},
        function() {console.log('error');}
    );
}
workOnLine();

complexLineTransformation = function (line) {
    var deferred = Q.defer();
    // ... async call goes here, in callback: deferred.resolve('done ok'); or deferred.reject(new Error(error));
    return deferred.promise;
}

function LineReader (filename) {      
  this.moreLinesAvailable = true;
  this.fd = undefined;
  this.bufferSize = 1024*1024;
  this.buffer = new Buffer(this.bufferSize);
  this.leftOver = '';

  this.read = undefined;
  this.idxStart = undefined;
  this.idx = undefined;

  this.lineNumber = 0;

  this._bundleOfLines = [];

  this.open = function() {
    this.fd = fs.openSync(filename, 'r');
  };

  this.readNextLine = function () {
    if (this._bundleOfLines.length === 0) {
      this._readNextBundleOfLines();
    }
    this.lineNumber++;
    var lineToReturn = this._bundleOfLines[0];
    this._bundleOfLines.splice(0, 1); // remove first element (pos, howmany)
    return lineToReturn;
  };

  this.getLineNumber = function() {
    return this.lineNumber;
  };

  this._readNextBundleOfLines = function() {
    var line = "";
    while ((this.read = fs.readSync(this.fd, this.buffer, 0, this.bufferSize, null)) !== 0) { // read next bytes until end of file
      this.leftOver += this.buffer.toString('utf8', 0, this.read); // append to leftOver
      this.idxStart = 0
      while ((this.idx = this.leftOver.indexOf("\n", this.idxStart)) !== -1) { // as long as there is a newline-char in leftOver
        line = this.leftOver.substring(this.idxStart, this.idx);
        this._bundleOfLines.push(line);        
        this.idxStart = this.idx + 1;
      }
      this.leftOver = this.leftOver.substring(this.idxStart);
      if (line !== "") {
        break;
      }
    }
  }; 
}

Я сделал модуль узла для асинхронного чтения большого файла текста или JSON. Проверено на больших файлах.

var fs = require('fs')
, util = require('util')
, stream = require('stream')
, es = require('event-stream');

module.exports = FileReader;

function FileReader(){

}

FileReader.prototype.read = function(pathToFile, callback){
    var returnTxt = '';
    var s = fs.createReadStream(pathToFile)
    .pipe(es.split())
    .pipe(es.mapSync(function(line){

        // pause the readstream
        s.pause();

        //console.log('reading line: '+line);
        returnTxt += line;        

        // resume the readstream, possibly from a callback
        s.resume();
    })
    .on('error', function(){
        console.log('Error while reading file.');
    })
    .on('end', function(){
        console.log('Read entire file.');
        callback(returnTxt);
    })
);
};

FileReader.prototype.readJSON = function(pathToFile, callback){
    try{
        this.read(pathToFile, function(txt){callback(JSON.parse(txt));});
    }
    catch(err){
        throw new Error('json file is not valid! '+err.stack);
    }
};

Просто сохраните файл как file-reader.js и используйте его следующим образом:

var FileReader = require('./file-reader');
var fileReader = new FileReader();
fileReader.readJSON(__dirname + '/largeFile.json', function(jsonObj){/*callback logic here*/});