批量修改elasticsearch的Nodejs流

Question

我想将下面的对象附加到流中的每个对象

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

我的流看起来像这样

[
  {"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"}
]

我的流应该是什么样子的！

> -> 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"}

如何在我现有的以下代码库中使用jsonstream来实现这一点？

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)
})

我目前使用的npm包

用于在elasticsearch中批量插入！

elasticsearch-writable-stream

用于从postgres获取数据到流中！

pg-query-stream

缺少的部分是将postgres流转换为弹性可写流！任何关于如何实现这一点的建议，指针，建议！

Answer 1

因此，基本上，不需要太多代码更改的唯一可行的选择是从postgres本身而不是node.js对象构建批量插入到elastic search所需的格式！

"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"