Node.js: Потоковая передача от MongoDB к файлу

Question

Я довольно новичок в javascript/node.js, и я пытаюсь получить очень простой сценарий работы: подключитесь к MongoDB, преобразуйте ответ JSON в CSV, напишите его в файл. Я пробовал как показано ниже:

fs = require('fs'); 
var MongoClient = require('mongodb').MongoClient; 
var Db = require('mongodb').Db; 
var Server = require('mongodb').Server; 
var Json2csvStream = require('json2csv-stream'); 
var Stream = require('stream'); 
var JSONStream = require('JSONStream'); 
var es = require('event-stream'); 
var csv = require('csv'); 

var fields = ['execAmendTime', 'execTime', 'execType', 'lastMkt', 'manualExecFlag', 'orderId', 'riskTrade', 'rootOrdId', 'salesCommissionRate', 'salesCommissionType', 'theoPov20Px', 
'theoPov20BL', 'tradeFlags', 'tradeNotes', 'transactTime', 'version', 'book.bookName', 'businessUnit', 'commissionRate', 'commissionSource', 'commissionType', 'counterBook.bookName', 
'counterParty.name', 'createTime', 'currency', 'direction', 'execQuantity', 'fxRate','orderQuantity', 'positionTrader.name', 'price', 'primaryTrader.name','rootSystem', 'source', 
'sectorGicsLevel1', 'salesTrader.name', 'tradedPrice', 'isCRB', 'clientCategory', 'tradeId', 'tradeDate', 'instrument.instrumentRic', 'notionalUSD','commissionUSD', 'region']; 

// Connect to the db 
MongoClient.connect("mongodb://*****", function (err, db) { 
    if (err) { return console.dir(err); } 

    if (!err) { 
     console.log("We are connected"); 
    } 

    db.open(function (err, db) { 
     if (err) { return console.dir(err); } 
     var newDb = db.db("test_db"); 

     var collection = newDb.collection('test', function (err, collection) { 
      if (err) { return console.dir(err); } 
      var parser = new Json2csvStream(); 
      var writer = fs.createWriteStream('out.csv'); 
      var stream = collection.find({ tradeDate: new Date('2015-12-29T00:00:00.000Z') }).stream(); 

      stream.pipe(parser).pipe(writer); 

      stream.on("data", function (item) { 
       console.log(item); 
      }); 

      stream.on('end', function() { 
       console.log("ended"); 
      }); 

      stream.on("end", function() { 
       newDb.close(); 
       db.close(); 
      }); 
     }); 
    }); 
}); 

Я получаю ошибки, как показано ниже.

Я пытался добавить преобразования с помощью JSON.stringify и т. Д., Но ни одна из моих попыток не работала. Кажется, мне нужно подождать, пока поток запросов к Mongo не завершится, прежде чем начать подавать его в трансформатор json2csv?

Любые идеи? Я что-то принципиально неправильно здесь?

Большое спасибо!

Выход:

We are connected 
D:\WebTrial\MongoProject\node_modules\mongodb\lib\utils.js:98 
    process.nextTick(function() { throw err; }); 
           ^

TypeError: Invalid non-string/buffer chunk 
    at validChunk (_stream_writable.js:178:14) 
    at Writable.write (_stream_writable.js:205:12) 
    at ondata (_stream_readable.js:525:20) 
    at emitOne (events.js:82:20) 
    at emit (events.js:169:7) 
    at readableAddChunk (_stream_readable.js:146:16) 
    at Readable.push (_stream_readable.js:110:10) 
    at D:\WebTrial\MongoProject\node_modules\mongodb\lib\cursor.js:1102:10 
    at handleCallback (D:\WebTrial\MongoProject\node_modules\mongodb\lib\utils.j 
s:96:12) 
    at D:\WebTrial\MongoProject\node_modules\mongodb\lib\cursor.js:673:5 

Answer 1

Это потому, что find().stream() потоки объектов в то время как Json2csvStream ожидает строки. event-stream может помочь вам укрепить объекты. Я также упростил ваш код, были ненужные вещи:

var fs = require('fs'); 
var MongoClient = require('mongodb').MongoClient; 
var es = require('event-stream'); 
var Json2csvStream = require('json2csv-stream'); 

// var Db = require('mongodb').Db; 
// var Server = require('mongodb').Server; 
// var Stream = require('stream'); 
// var JSONStream = require('JSONStream'); 
// var csv = require('csv'); 

var fields = ['execAmendTime', 'execTime', 'commissionUSD', 'region']; 

// Connect to the db 
// you can put the db name in the url 
MongoClient.connect("mongodb://localhost:27017/test_db", function (err, db) { 
    if (err) { 
     return console.dir(err); 
    } else { 
     console.log("We are connected"); 
    } 

    // without strict: true, err is always null 
    // in strict mode, there is an err if the collection doesn't exist 
    db.collection('test', { strict: true }, function (err, collection) { 
     if (err) { 
      return console.dir(err); 
     } 

     var json2csv = new Json2csvStream(); 
     var writer = fs.createWriteStream('out.csv'); 

     var mongoStream = collection.find(
      { tradeDate: new Date('2015-12-29T00:00:00.000Z') } 
     ).stream(); 

     var stream = mongoStream 
      .pipe(es.map(function (doc, next) { 
       doc = JSON.stringify(doc); 
       // console.log(doc); 
       next(null, doc); 
      })).pipe(json2csv).pipe(writer).on('close', function() { 
       console.log('done...'); 
       db.close(); 
      }); 
    }); 
});