NodeJS streamen Modifikation für Elasticsearch bulk

Question

i unten Objekt zu jedem Objekt in Strom

{"index":{"_index":"tvseries","_type":"internindex"}} 

meinem Stream sieht wie folgt aus

[ 
    {"showname":"The X Files","episode":"04","content":"Before what?","season":"1"}, 
    {"showname":"The X Files","episode":"04","content":"Before what?","season":"1"}, 
    {"showname":"The X Files","episode":"01","content":"What?","season":"1"} 
] 

was mein Strom aussehen sollte anhängen wollte!

> -> POST http://localhost:9200/_bulk {"index":{"_index":"tvseries","_type":"internindex"}} 
> {"showname":"The X Files","episode":"04","content":"Before 
> what?","season":"1"} 
> {"index":{"_index":"tvseries","_type":"internindex"}} 
> {"showname":"The X 
> Files","episode":"04","content":"Great.","season":"1"} 
> {"index":{"_index":"tvseries","_type":"internindex"}} 
> {"showname":"The X 
> Files","episode":"01","content":"What?","season":"1"} 

, wie ich unter Code-Basis in meinem bestehenden

var stream = new ElasticsearchWritableStream(client, { 
    highWaterMark: 256, 
    flushTimeout: 500 
}); 

pg.connect(connectionString,function(err, client, done) { 
    if(err) throw err; 
    var query = new QueryStream('SELECT * FROM srt limit 2') 
    var streams = client.query(query) 

    //release the client when the stream is finished 
    streams.on('end', done) 
    streams.pipe(JSONStream.stringify()).pipe(stream) 
}) 

die npm Pakete dies mit jsonstream erreichen kann ich zur Zeit

in Elasticsearch Für Groß Einfügung bin mit!

elasticsearch-writable-stream

für Daten von Postgres in Ströme bekommen!

pg-query-stream

fehlende Stück wandelt Postgres Ströme in elastischen beschreibbaren Ströme! irgendein Vorschlag, Hinweise, Empfehlungen, wie man das erreicht!

Answer 1

so dass im Grunde die einzig gangbare Optionen ohne wäre viel Codeänderung sein Format für Massen in Elasticsearch Einsatz selbst von Postgres erforderlich bauen anstatt in node.js Objekt!

"SELECT 'tvseries' as index,'internindex' as type, json_build_object('showname', showname, 'epsiode', ep,'content',content,'season',season) AS body" 
+" FROM srt where shownameid=4"