elasticsearch - logstash 输出到 elasticsearch 索引和映射

Question

我正在尝试将 logstash 输出到 elasticsearch，但我不确定如何使用我在 elasticsearch 中定义的映射...

在 Kibana 中，我是这样做的:

像这样创建了一个索引和映射:

PUT /kafkajmx2
{
  "mappings": {
    "kafka_mbeans": {
      "properties": {
        "@timestamp": {
          "type": "date"
        },
        "@version": {
          "type": "integer"
        },
        "host": {
          "type": "keyword"
        },
        "metric_path": {
          "type": "text"
        },
        "type": {
          "type": "keyword"
        },
        "path": {
          "type": "text"
        },
        "metric_value_string": {
          "type": "keyword"
        },
        "metric_value_number": {
          "type": "float"
        }
      }
    }
  }

}

可以像这样向它写入数据:

POST /kafkajmx2/kafka_mbeans
{
  "metric_value_number":159.03478490788203,
  "path":"/home/usrxxx/logstash-5.2.0/bin/jmxconf",
  "@timestamp":"2017-02-12T23:08:40.934Z",
  "@version":"1","host":"localhost",
  "metric_path":"node1.kafka.server:type=BrokerTopicMetrics,name=TotalFetchRequestsPerSec.FifteenMinuteRate",
  "type":null


}

现在我的 logstash 输出如下所示:

input {
        kafka {
                kafka details here
        }

}
output {

    elasticsearch {
            hosts => "http://elasticsearch:9050"
            index => "kafkajmx2"

    }

}

它只是将它写入 kafkajmx2 索引，但不使用 map ，当我在 kibana 中这样查询时:

get /kafkajmx2/kafka_mbeans/_search?q=*
{


}

我得到了这个:

      {
        "_index": "kafkajmx2",
        "_type": "logs",
        "_id": "AVo34xF_j-lM6k7wBavd",
        "_score": 1,
        "_source": {
          "@timestamp": "2017-02-13T14:31:53.337Z",
          "@version": "1",
          "message": """
{"metric_value_number":0,"path":"/home/usrxxx/logstash-5.2.0/bin/jmxconf","@timestamp":"2017-02-13T14:31:52.654Z","@version":"1","host":"localhost","metric_path":"node1.kafka.server:type=SessionExpireListener,name=ZooKeeperAuthFailuresPerSec.Count","type":null}

"""
        }
      }

我如何告诉它在 logstash 输出中使用 map kafka_mbeans？

-----编辑-----

我试过这样的输出，但仍然得到相同的结果:

output {

        elasticsearch {
                hosts => "http://10.204.93.209:9050"
                index => "kafkajmx2"
                template_name => "kafka_mbeans"
                codec => plain {
                        format => "%{message}"
                }

        }

}

elasticsearch中的数据应该是这样的:

{
  "@timestamp": "2017-02-13T14:31:52.654Z", 
  "@version": "1", 
  "host": "localhost", 
  "metric_path": "node1.kafka.server:type=SessionExpireListener,name=ZooKeeperAuthFailuresPerSec.Count", 
  "metric_value_number": 0, 
  "path": "/home/usrxxx/logstash-5.2.0/bin/jmxconf", 
  "type": null
}

--------编辑 2----------------

我至少通过添加这样的过滤器得到了要解析为 json 的消息:

input {
        kafka {
                ...kafka details....
        }

}
filter {
        json {
                source => "message"
                remove_field => ["message"]
        }
}
output {

        elasticsearch {
                hosts => "http://node1:9050"
                index => "kafkajmx2"
                template_name => "kafka_mbeans"
        }

}

它仍然不使用模板，但至少可以正确解析 json...所以现在我明白了:

  {
    "_index": "kafkajmx2",
    "_type": "logs",
    "_id": "AVo4a2Hzj-lM6k7wBcMS",
    "_score": 1,
    "_source": {
      "metric_value_number": 0.9967205071482902,
      "path": "/home/usrxxx/logstash-5.2.0/bin/jmxconf",
      "@timestamp": "2017-02-13T16:54:16.701Z",
      "@version": "1",
      "host": "localhost",
      "metric_path": "kafka1.kafka.network:type=SocketServer,name=NetworkProcessorAvgIdlePercent.Value",
      "type": null
    }
  }

Answer 1

你需要改变的很简单。首先在您的 kafka 输入中使用 json 编解码器。不需要 json 过滤器，您可以将其删除。

    kafka {
            ...kafka details....
            codec => "json"
    }

然后在您的 elasticsearch 输出中您缺少映射类型(下面的参数 document_type)，这很重要，否则它默认为 logs (如您所见)并且与您的 kafka_mbeans 映射类型不匹配。此外，您实际上并不需要使用模板，因为您的索引已经存在。进行如下修改:

    elasticsearch {
            hosts => "http://node1:9050"
            index => "kafkajmx2"
            document_type => "kafka_mbeans"
    }