NodeJS，promise，streams - 處理大型CSV文件

Question

我需要構建一個處理大型CSV文件以用於bluebird.map（）調用的函數。鑑於文件的可能大小，我想使用流式傳輸。

該函數應該接受一個流（一個CSV文件）和一個函數（處理來自流的塊），並在文件讀取結束（解析）或錯誤（拒絕）時返回一個承諾。

於是，我開始：

'use strict'; 

var _ = require('lodash'); 
var promise = require('bluebird'); 
var csv = require('csv'); 
var stream = require('stream'); 

var pgp = require('pg-promise')({promiseLib: promise}); 

api.parsers.processCsvStream = function(passedStream, processor) { 

    var parser = csv.parse(passedStream, {trim: true}); 
    passedStream.pipe(parser); 

    // use readable or data event? 
    parser.on('readable', function() { 
    // call processor, which may be async 
    // how do I throttle the amount of promises generated 
    }); 

    var db = pgp(api.config.mailroom.fileMakerDbConfig); 

    return new Promise(function(resolve, reject) { 
    parser.on('end', resolve); 
    parser.on('error', reject); 
    }); 

} 

現在，我有兩個相互關聯的問題：

1. 我需要節制的實際數據量進行處理，從而不會產生內存壓力。
2. 作爲processor參數傳遞的函數通常是異步的，例如通過基於promise的庫（現在：pg-promise）將文件的內容保存到數據庫。因此，它會在記憶中創造一個承諾並繼續前進。

的pg-promise圖書館功能來管理這一點，像page()，但我不能換我圍​​繞如何流事件處理程序與這些承諾搭配方式前進。現在，我在readable部分的每個read()之後返回承諾，這意味着我創建了大量承諾的數據庫操作，並最終出錯，因爲我遇到了進程內存限制。

有沒有人有這樣的工作示例，我可以用作跳點？

UPDATE：可能對皮膚貓不止一種方法，但這個工程：

'use strict'; 

var _ = require('lodash'); 
var promise = require('bluebird'); 
var csv = require('csv'); 
var stream = require('stream'); 

var pgp = require('pg-promise')({promiseLib: promise}); 

api.parsers.processCsvStream = function(passedStream, processor) { 

    // some checks trimmed out for example 

    var db = pgp(api.config.mailroom.fileMakerDbConfig); 
    var parser = csv.parse(passedStream, {trim: true}); 
    passedStream.pipe(parser); 

    var readDataFromStream = function(index, data, delay) { 
    var records = []; 
    var record; 
    do { 
     record = parser.read(); 
     if(record != null) 
     records.push(record); 
    } while(record != null && (records.length < api.config.mailroom.fileParserConcurrency)) 
    parser.pause(); 

    if(records.length) 
     return records; 
    }; 

    var processData = function(index, data, delay) { 
    console.log('processData(' + index + ') > data: ', data); 
    parser.resume(); 
    }; 

    parser.on('readable', function() { 
    db.task(function(tsk) { 
     this.page(readDataFromStream, processData); 
    }); 
    }); 

    return new Promise(function(resolve, reject) { 
    parser.on('end', resolve); 
    parser.on('error', reject); 
    }); 
} 

有人看到一個潛在的問題，這種做法？

Answer 1

查找下面的正確執行相同類型的任務的完整應用程序：它將文件讀取爲流，將其解析爲CSV並將每行插入到數據庫中。

const fs = require('fs'); 
const promise = require('bluebird'); 
const csv = require('csv-parse'); 
const pgp = require('pg-promise')({promiseLib: promise}); 

const cn = "postgres://postgres:password@localhost:5432/test_db"; 
const rs = fs.createReadStream('primes.csv'); 

const db = pgp(cn); 

function receiver(_, data) { 
    function source(index) { 
     if (index < data.length) { 
      // here we insert just the first column value that contains a prime number; 
      return this.none('insert into primes values($1)', data[index][0]); 
     } 
    } 

    return this.sequence(source); 
} 

db.task(t => { 
    return pgp.spex.stream.read.call(t, rs.pipe(csv()), receiver); 
}) 
    .then(data => { 
     console.log('DATA:', data); 
    } 
    .catch(error => { 
     console.log('ERROR:', error); 
    }); 

注意，我改變的唯一事情：利用圖書館csv-parse代替csv，作爲一個更好的選擇。

從spex庫中增加使用方法stream.read，它正確地服務於Readable流以用於承諾。

Answer 2

那麼說，你不想流式傳輸，但某種數據塊？ ;-)

你知道嗎https://github.com/substack/stream-handbook？

我認爲最簡單的方法不改變你的架構會是某種承諾池。例如https://github.com/timdp/es6-promise-pool

Answer 3

你可能想看看promise-streams

var ps = require('promise-streams'); 
passedStream 
    .pipe(csv.parse({trim: true})) 
    .pipe(ps.map({concurrent: 4}, row => processRowDataWhichMightBeAsyncAndReturnPromise(row))) 
    .wait().then(_ => { 
    console.log("All done!"); 
    }); 

工程與背壓和一切。