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索引但不使用地图,当我在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}

"""
        }
      }

如何告诉它kafka_mbeans在logstash输出中使用地图？

- - -编辑 - - -

我尝试了这样的输出,但仍然得到相同的结果:

output {

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

        }

}

弹性搜索中的数据应如下所示:

{
  "@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

你需要改变的是非常简单的.首先json在kafka输入中使用编解码器.无需json过滤器,您可以将其删除.

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

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

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