将 JSON 数据从 Kinesis Firehose 发送到 elasticsearch 无法识别实际值答案

【问题标题】：Real value not recognized sending JSON data from Kinesis Firehose to elasticsearch将 JSON 数据从 Kinesis Firehose 发送到 elasticsearch 无法识别实际值
【发布时间】：2017-05-09 07:35:21
【问题描述】：

我在 Kibana 中遇到了 value 字段的问题，下面几行对此进行了解释。我会试着解释一下情况。

我将 dynamoDB 流发送到 Lambda，然后发送到 Kenesis Firehouse，最后从 Firehose 发送到 Elasticsearch。我正在使用 Kibana 可视化数据，这就是我遇到问题的地方。

假设我将此 JSON 发送到 DynamoDB：

{
    "id": "identificator",
    "timestamp": "2017-05-09T06:38:00.337Z",
    "value": 33,
    "units": "units",
    "description": "This is the description",
    "machine": {
        "brand": "brand",
        "application": "application"
    }
}

在 Lambda 中，我收到以下信息：

{
    "data": {
        "M": {
            "machine": {
                "M": {
                    "application": {
                        "S": "application"
                    },
                    "brand": {
                        "S": "band"
                    }
                }
            },
            "description": {
                "S": "This is the description"
            },
            "id": {
                "S": "identificator"
            },
            "units": {
                "S": "units"
            },
            "value": {
                "N": "33"
            },
            "_msgid": {
                "S": "85209b75.f51ee8"
            },
            "timestamp": {
                "S": "2017-05-09T06:38:00.337Z"
            }
        }
    },
    "id": {
        "S": "85209b75.f51ee8"
    }
}

如果我将最后一个 JSON 转发到 Kinesis Firehose，当我在 Kibana 中配置索引模式时，它会自动识别 "timestamp"（这很棒）。这里的问题是，"value" 字段就像一个字符串，无法识别。

我尝试修改 JSON，然后将其再次发送到 Firehose，但 Kibana 无法识别 "timestamp"：

{
    "data": {
        "machine": {
            "application": "application",
            "brand": "brand"
        },
        "description": "This is the description",
        "id": "identificator",
        "units": "KWh",
        "value": 33,
        "_msgid": "85209b75.f51ee8",
        "timestamp": "2017-05-09T06:38:00.337Z"
    },
    "id": "85209b75.f51ee8"
}

我想知道如何发送这些数据，并且 Kibana 可以识别“时间戳”和“值”字段。

这是我在 lambda 中使用的代码示例：

var AWS = require('aws-sdk');
var unmarshalJson = require('dynamodb-marshaler').unmarshalJson;

var firehose = new AWS.Firehose();

exports.lambda_handler = function(event, context) {

    var record = JSON.stringify(event.Records[0].dynamodb.NewImage);

    console.log("[INFO]:"+JSON.stringify(event.Records[0].dynamodb.NewImage));

    var params = {
        DeliveryStreamName: 'DeliveryStreamName',

        Record:{ 
            Data: record
        }
    };
    firehose.putRecord(params, function(err, data) {

        if (err) console.log(err, err.stack); // an error occurred
        else     console.log(JSON.stringify(data));           // successful response

        context.done();
    });
};

【问题讨论】：

标签： json elasticsearch lambda kibana amazon-dynamodb-streams

【解决方案1】：

我解决了它自己创建索引映射而不是让 Kinesis Firehose 创建它。并将"timestamp" 属性声明为{ "type" : "date" }，将"value" 属性声明为{ "type" : "float" }

例如对于这种类型的 JSON：

{
    "data": {
        "timestamp": "2017-05-09T11:30:41.484Z",
        "tag": "tag",
        "value": 33,
        "units": "units",
        "type": "type",
        "machine":{
            "name": "name",
            "type": "type",
            "company": "company"
        }
    },
    "id": "85209b75.f51ee8"
}

我手动创建了以下 elasticsearch 索引和映射：

PUT /index
{
    "settings" : {
        "number_of_shards" : 2
    },
    "mappings" : {
        "type" : {
            "properties" : {
                "data" : {
                    "properties" : {
                        "machine":{
                            "properties": {
                                "name": { "type" : "text" },
                                "type": { "type" : "text" },
                                "company": { "type" : "text" }
                            }
                        },
                        "timestamp": { "type" : "date" },
                        "tag" : { "type" : "text" },
                        "value": { "type" : "float" },
                        "description":  { "type" : "text" },
                        "units":  { "type" : "text" },
                        "type" : { "type" : "text" },
                        "_msgid":  { "type" : "text" }
                    }
                },
                "id":  { "type" : "text" }      
            }
        }
    }
}

所以，要解决它，我认为更好的解决方案是在 lambda 中您必须检查索引映射是否存在，如果不存在则自行创建。

【讨论】：