javascript - Node - 无数据创建的 ES 索引

Question

我一直在开发一个 AWS Lambda 函数，用于在每次从 Cloudfront 接收日志时对 S3 存储桶发送的通知使用react。 我能够解压缩日志包并使用 Cloudfront Parser 解析它们。

但是，我仍然无法将它们发送到 ElasticSearch 端点。我主要使用 winston-elasticsearch和 s3-to-logstore但他们都没有工作。

这是我的代码:

exports.handler = function(event, context, callback) {
    var srcBucket = event.Records[0].s3.bucket.name;
    var srcKey = event.Records[0].s3.object.key;

    async.waterfall([
        function fetchLogFromS3(next){
            console.log('Fetching compressed log from S3...');
            s3.getObject({
               Bucket: srcBucket,
               Key: srcKey
            },
            next);
        },
        function uncompressLog(response, next){
            console.log("Uncompressing log...");
            zlib.gunzip(response.Body, next);
        },
        function publishNotifications(jsonBuffer, next) {
            console.log('Filtering log...');
            var json = jsonBuffer.toString();
            console.log('CloudFront JSON from S3:', json);

            var records;
            CloudFrontParser.parse(json, { format: 'web' }, function (err, accesses) {
              if(err){
                console.log(err);
              } else {
                records = accesses;
              }
            });

            // Here, how to send the parsed data? 

            console.log('CloudFront parsed:', records);
        }
    ], function (err) {
        if (err) {
            console.error('Failed to send data: ', err);
        } else {
            console.log('Successfully send data.');
        }
        callback(null,"message");
    });
};

有没有简单的方法向ES发送数据？
类似于:

var client = new elasticsearch.Client({
  host: process.env.ES_HOST,
  log: 'trace',
  keepAlive: false
});

client.index({
        index: 'cloudfront_index',
        type: 'log',
        body: records
        }, function(err, resp, status) {
        console.log(resp);
});

它可以工作，但不会发送数据:

GET cloudfront_index/_search
{
  "took": 0,
  "timed_out": false,
  "_shards": {
    "total": 5,
    "successful": 5,
    "skipped": 0,
    "failed": 0
  },
  "hits": {
    "total": 0,
    "max_score": null,
    "hits": []
  }
}

Answer 1

你就快到了。您需要使用 bulk method实现你想要的:

var client = new elasticsearch.Client({
  host: process.env.ES_HOST,
  log: 'trace',
  keepAlive: false
});

var bulk = [];
records.forEach(function(record) {
    bulk.push({"index": {}})
    bulk.push(record);
});
client.bulk({
        index: 'cloudfront_index',
        type: 'log',
        body: bulk
}, function(err, resp, status) {
        console.log(resp);
});