Удалить поле «сообщение» для данных с метрическими битами

Question

Я пытаюсь получить данные metricbeat, используя logstash. (Kafka как вход & ElasticSearch как выход).

input { 
    kafka { 
    bootstrap_servers => "XX.XX.XXX.XX:9092" 
    topics => ["cc-data"] 
    } 
} 
output { 
    elasticsearch { 
    hosts => ["XX.XX.XXX.XX:9200"] 
    index => "metricbeat-%{+YYYY.MM.dd}" 
    } 
} 

В выходе я вижу данные, как:

 { 
     "_index" : "metricbeat-2017.01.18", 
     "_type" : "logs", 
     "_id" : "AVmzQUytyldhSr4kUaUS", 
     "_score" : 1.0, 
     "_source" : { 
      "@timestamp" : "2017-01-18T20:21:45.324Z", 
      "@version" : "1", 
      "message" : "{\"@timestamp\":\"2017-01-18T20:21:44.394Z\",\"beat\":{\"hostname\":\"ip-XX-XXX-XX-XXX\",\"name\":\"ip-XX-XXX-XX-XXX\",\"version\":\"5.0.0\"},\"metricset\":{\"module\":\"system\",\"name\":\"process\",\"rtt\":104061},\"system\":{\"process\":{\"cpu\":{\"start_time\":\"2016-08-03T10:10:17.000Z\",\"total\":{\"pct\":0.000000}},\"fd\":{\"limit\":{\"hard\":4096,\"soft\":1024},\"open\":0},\"memory\":{\"rss\":{\"bytes\":0,\"pct\":0.000000},\"share\":0,\"size\":0},\"name\":\"migration/2\",\"pgid\":0,\"pid\":47,\"ppid\":2,\"state\":\"sleeping\",\"username\":\"root\"}},\"type\":\"metricsets\"}" 
     } 
     } 
    ] 
    } 
} 

Я не хочу «сообщение» в одиночку FIELD .. Вместо этого, все Данные в этой области должны появиться, как это [хочу удалите только имя поля «сообщение», но содержимое этого должно появиться] Как удалить имя поля в отдельности и сохранить содержимое как есть.

Может ли кто-нибудь помочь, как мы это достигнем?

Answer 1

По умолчанию вход kafka имеет кодек plain, что означает, что он использует тему и интерпретирует контент как обычную строку. Что вам нужно, вместо того, чтобы изменить кодек json, чтобы он разбирает содержимое в формате JSON структуры:

input { 
    kafka { 
    bootstrap_servers => "XX.XX.XXX.XX:9092" 
    topics => ["cc-data"] 
    codec => "json"   # <---- add this 
    } 
}