弹性搜索批量的Nodejs流修改答案

【问题标题】：Nodejs stream modification for elasticsearch bulk弹性搜索批量的Nodejs流修改
【发布时间】：2017-12-17 21:01:46
【问题描述】：

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

{"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-writable-stream

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

pg-query-stream

缺少的部分是将 postgres 流转换为弹性可写流！关于如何实现这一目标的任何建议、指示和建议！

【问题讨论】：

标签： node.js postgresql elasticsearch node-postgres elasticsearch.js

【解决方案1】：

所以基本上，没有太多代码更改的唯一可行选项是构建从 postgres 本身而不是在 node.js 对象中批量插入弹性搜索所需的格式！

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

【讨论】：