HTTP POST 弹性搜索事件流批量答案

【问题标题】：HTTP POST elastic search event stream bulkHTTP POST 弹性搜索事件流批量
【发布时间】：2014-12-04 00:43:36
【问题描述】：

我有一个 node.js 程序，它使用流来读取文件 (nodejs event stream setting a variable per stream)

我想使用相同的程序将此数据写入弹性搜索。我写了一个小写函数

var writeFunction = function(data) {
    //console.log(data);
    var client = request.newClient("http://localhost:9200");
    client.post('/newtest3/1',data,function(err,res,body) {
        return console.log(res.statusCode);
    });
};

并将其与流媒体联系起来

var processMyFile = function(file) {
    var stream = getStream(file);
    var nodeName = stream.nodeName;
    stream
        .pipe(es.split())
        .on('end',endFunction)
        .pipe(es.map(function(data,cb) {
            processFunction(nodeName,data,cb);
        }))
        .pipe(es.map(function(data,cb) {
            writeFunction(data);
        }));

}

以上内容按预期异步工作并写入数据，除了它需要很长时间。它似乎也可以作为缓冲区工作，因为写入需要比读取更长的时间。（使用管道的优势）我知道弹性搜索中有一个批量接口，我可以使用它导入。 Kibana 入门指南中的shakesphere.json 示例 (http://www.elasticsearch.org/guide/en/kibana/current/using-kibana-for-the-first-time.html)

这意味着我需要以批量导入所需的格式创建文件，然后运行 curl 程序等。我想避免创建临时文件。

作为流式处理的一部分，有没有更简单的方法可以更快地将数据导入到 elasticsearch 中

【问题讨论】：

标签： node.js asynchronous elasticsearch stream event-stream

【解决方案1】：

elasticsearch-streams 将帮助您使用带流式传输的批量接口，无需先编写 json 文件。

我相信你的代码或多或少是这样的：

var TransformToBulk = require('elasticsearch-streams').TransformToBulk
var WritableBulk = require('elasticsearch-streams').WritableBulk;
var client = new require('elasticsearch').Client();

var bulkExec = function(bulkCmds, callback) {
  client.bulk({
    index : 'newtest3',
    type  : '1',
    body  : bulkCmds
  }, callback);
};

var ws = new WritableBulk(bulkExec);
var toBulk = new TransformToBulk(function getIndexTypeId(doc) { return { _id: doc.id }; });

var processMyFile = function(file) {
  var stream = getStream(file);

  stream
    .pipe(toBulk)
    .pipe(ws)
    .on('close', endFunction)
    .on('err', endFunction);
}

【讨论】：