内存耗尽:在大数据处理中终止进程错误

Question

我有几个.csv文件，需要与另一个大型.csv文件(超过30万行)进行比较，并且在服务器上遇到内存不足的错误。我正在一个4GB内存的服务器上运行这个程序，所以我不知道为什么会发生这种情况，但是我的代码如下所示。我正在使用ya-csv在csv行中阅读：

var csv = require('ya-csv');
var fs = require('graceful-fs');
var async = require('async');


var first_silo = [];
var second_Silo = [];
var combined = [];

var reader = csv.createCsvFileReader('december_raw.csv', {columnsFromHeader:true,'separator': ','});
var first = csv.createCsvFileReader('first_data.csv', {columnsFromHeader:false,'separator': ','});
var second = csv.createCsvFileReader('second_data.csv', {columnsFromHeader:false,'separator': ','})


async.series([
 //push data from other .csv files into arrays
function(callback){
   first.addListener('data', function(data){
      first_silo.push(data[0]);
   })
   first.addListener('end', function(){
      callback();
   })
},

function(callback){
   second.addListener('data', function(data){
       second_silo.push(data[0]);
   });
   second.addListener('end', function(data){
       callback();
   });
},

function(callback){
    reader.addListener('data', function(data){
       //compare the data from reader to each item in the first array and append the items that get a match to a .csv.
       for(var i=0;i<first_silo.length;i++){
           if(data[0] === first_silo[i]){
               fs.appendFileSync('results.csv', data[0]+","+first_silo[i])
               break;
           }
       } 
    });
},

function(callback){
    reader.addListener('data', function(data){
        //do the same with the first array as the second.
        for(var i=0;i<second_silo.length;i++){
            if(data[0] === second_silo[i]){
               fs.appendFileSync('results.csv', data[0]+","+second_silo[i]);
               break;
            }
        }
    })
}
])

当我这样做的时候，我就不知道过去的first_silo比较了。节点应用程序将停止，当我使用dmesg时，我可以看到内存不足的错误。

我也尝试用这个标志运行这个程序：

-最大-旧-空间大小=3000

我还是会犯同样的错误。

有更聪明的方法吗？任何帮助都将不胜感激。

Answer 1

这里有一个更有记忆效率的答案，没有任何假设。在它中，确保将最小的CSV文件作为第一个参数传递给compareRows函数。

这确实确保了您的内存效率尽可能高，只将最小的集合存储在内存中。

var csv = require('ya-csv');
var fs = require('graceful-fs');

var smallFileName = ""; // used to see if we need to really reload the file again.
var smaller_silo = [];

compareRows('smaller.csv', 'larger.csv', function(){
  compareRows('smaller.csv', 'anotherLarger.csv', function(){
    smaller_silo = []; }); // done
});

function compareRows(smallerFileName, largerFileName, callBack){

  var reader;
  if(smallerFileName !== smallFileName){
    smallFileName = smallerFileName;
    reader = csv.createCsvFileReader(smallerFileName, { columnsFromHeader: true, separator: ','});
    reader.addListener('data', function(data){
      smaller_silo.push(data[0]);
    });

    reader.addListener('end', function(){
      compareSmallerToLarger(largerFileName, callBack);
    });
  }
  else{
    compareSmallerToLarger(largerFileName, callBack);
  }
}

function compareSmallerToLarger(largerFileName, callBack){

  var csvStream = csv.createCsvFileReader( largerFileName, { columnsFromHeader: false, 'separator':','});
  csvStream.addListener('data', function(data){
    for (var i = 0; i < smaller_silo.length; i++) {
      if(data[0] === smaller_silo[i]){
        fs.appendFileSync('results.csv', data[0]+","+smaller_silo[i]);
        break;
      }
    }
  });
  csvStream.addListener('end', function(data){
    if(callBack && typeof callBack === "function") callBack();
  });
}

不管怎么说，我不应该纠结于事情.

Answer 2

由于几个原因，您的算法运行效率很低。请原谅我，但是我将在不使用您正在使用的async.series调用的情况下这样做。希望它仍然是有用的。

第一件事是第一件事:我在做一个假设。我假设第一个文件december_raw.csv的数据大小小于第二个和第三个文件。即使不是这样，只要文件的内容没有超出内存限制，这仍然可以工作，而不会耗尽内存。

其次，您正在同时加载两个数组，而不是一次加载一个数组。这基本上使你的内存使用量翻了一番。

第三，我的预感是，当您运行csv.createCsvFileReader时，您将同时开始对它们进行流处理。你可能不想要这个。

因为您要将两个文件与december_raw.csv的内容进行比较，所以最好将该文件的内容完全加载到内存中，然后再流--使用callBack和通用比较函数将其他两个文件串联起来进行比较。

var csv = require('ya-csv');
var fs = require('graceful-fs');

var reader_silo = []; // a variable that holds the rows of the main csv.

var reader = csv.createCsvFileReader('december_raw.csv', {columnsFromHeader:true,'separator': ','});
reader.addListener('data', function(data){
  reader_silo.push(data[0]); // load each read in row into the array
});

reader.addListener('end', function(){
  //start comparing with first csv file.
  compareRows('first_data.csv', function(){
    // compare with second data
    compareRows('second_data.csv');
  });
});

// the comparison function, takes in the filename, and a callBack if there is one.    
function compareRows(csvFileName, callBack){

  var csvStream = csv.createCsvFileReader(csvFileName, {columnsFromHeader:false,'separator': ','}); // begin stream

  csvStream.addListener('data', function(data){
    for (var i = 0; i < reader_silo.length; i++) {
      if(data[0] === reader_silo[i]){
        fs.appendFileSync('results.csv', data[0]+","+reader_silo[i]);
        break;
      }
    }
  });

  csvStream.addListener('end', function(data){
    // if there's a callBack then we can execute it.
    // in this case the first time it is executed there is a callBack which executes this function again with the next file.
    if(callBack && typeof callBack === "function") callBack();
  });
}

PS。如果您的脚本继续下去，那么在进行比较时，您可能还需要考虑将reader_silo归零。因此，您的'end'侦听器callBack应该如下所示：

reader.addListener('end', function(){
  compareRows('first_data.csv', function(){
    compareRows('second_data.csv', function(){
      reader_silo = [];
    });
  });
});