通过使用 kafka 的 elasticsearch 使时间戳字段在 kibana 中可用答案

【问题标题】：Making timestamp field available in kibana via elasticsearch with kafka通过使用 kafka 的 elasticsearch 使时间戳字段在 kibana 中可用
【发布时间】：2019-12-10 06:39:03
【问题描述】：

通过 Kafka elasticsearch 连接器添加数据时，如何允许 kibana 使用我的时间戳字段作为@timestamp？

我正在像这样定义我的 avro 架构

    public static String userSchema =  "{\"type\":\"record\"," +
                                         "\"name\":\"myrecord\"," +
                                         "\"fields\":[" +
                                         "{\"name\":\"wSrcTime\",\"type\":[\"string\", \"null\"],\"default\":\"null\"}," +
                                         "{\"name\":\"wTradePrice\",\"type\":[\"null\",\"float\"],\"default\":null}," +
                                         "{\"name\":\"timestamp\",\"type\":{\"type\":\"long\",\"logicalType\":\"timestamp-millis\"}}" +
                                         "]}";

并使用它来填充字段

avroRecord.put("timestamp", System.currentTimeMillis());

我看到kafka-avro-console-consumer的数据如下：

{"wSrcTime":{"string":"2019-08-01 15:20:40.127"},"wTradePrice":null,"timestamp":1564672840137}
{"wSrcTime":{"string":"2019-08-01 15:20:41.062"},"wTradePrice":null,"timestamp":1564672841072}
{"wSrcTime":{"string":"2019-08-01 15:20:41.062"},"wTradePrice":null,"timestamp":1564672841073}
{"wSrcTime":{"string":"2019-08-01 15:20:41.064"},"wTradePrice":null,"timestamp":1564672841075}
{"wSrcTime":{"string":"2019-08-01 15:20:41.065"},"wTradePrice":null,"timestamp":1564672841076}
{"wSrcTime":{"string":"2019-08-01 15:20:41.410"},"wTradePrice":null,"timestamp":1564672841420}

并将添加到 kibana 索引的数据视为

timestamp: number
wTradePrice: number
wSrcTime: string

有没有推荐的方法来制作时间戳@timestamp，以便我可以在轴上使用它？谢谢

【问题讨论】：

标签： kibana avro apache-kafka-connect kafka-producer-api

【解决方案1】：

您可以在 elasticsearch 上创建一个摄取管道，以便在索引文档之前将时间戳字段重命名为 @timestamp。

在弹性上创建管道：

PUT _ingest/pipeline/rename_timestamp
{
  "rename": {
    "field": "timestamp",
    "target_field": "@timestamp"
  }
}

并且您应该在将新文档索引到数据库时提供管道名称，例如

PUT /es-index/_doc?pipeline=rename_timestamp
{
  "timestamp": "value"
  ...
}

【讨论】：