logstash 5.0.1：为多个 kafka 输入主题设置弹性搜索多个索引输出答案

【问题标题】：logstash 5.0.1: setup elasticsearch multiple indexes ouput for multiple kafka input topicslogstash 5.0.1：为多个 kafka 输入主题设置弹性搜索多个索引输出
【发布时间】：2017-01-23 18:06:10
【问题描述】：

我有一个logstash输入设置为

input {
  kafka {
  bootstrap_servers => "zookeper_address"
  topics => ["topic1","topic2"]
  }
}

我需要将主题输入到 elasticsearch 中的两个不同索引中。任何人都可以帮助我如何为此类任务设置输出。目前我只能设置

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "my_index"
    codec => "json"
    document_id => "%{id}"
  }
}

我需要在同一个 elasticsearch 实例上使用两个索引，例如 index1 和 index2，这将由来自 topic1 和 topic2 的消息提供。

【问题讨论】：

标签： elasticsearch logstash-configuration

【解决方案1】：

首先，您需要将decorate_events 添加到您的kafka 输入中，以便了解消息来自哪个主题

input {
  kafka {
    bootstrap_servers => "zookeper_address"
    topics => ["topic1","topic2"]
    decorate_events => true
  }
}

然后，您有两个选择，都涉及条件逻辑。首先是通过引入一个过滤器来根据主题名称添加正确的索引名称。为此，您需要添加

filter {
   if [kafka][topic] == "topic1" {
      mutate {
         add_field => {"[@metadata][index]" => "index1"}
      }
   } else {
      mutate {
         add_field => {"[@metadata][index]" => "index2"}
      }
   }
   # remove the field containing the decorations, unless you want them to land into ES
   mutate {
      remove_field => ["kafka"]
   }
}
output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "%{[@metadata][index]}"
    codec => "json"
    document_id => "%{id}"
  }
}

然后第二个选项是直接在输出部分执行 if/else，像这样（但额外的 kafka 字段将进入 ES）：

output {
   if [@metadata][kafka][topic] == "topic1" {
     elasticsearch {
       hosts => ["localhost:9200"]
       index => "index1"
       codec => "json"
       document_id => "%{id}"
     }
   } else {
     elasticsearch {
       hosts => ["localhost:9200"]
       index => "index2"
       codec => "json"
       document_id => "%{id}"
     }
   }
}

【讨论】：

感谢您详尽的回复。非常感谢
那么你能做些什么来使用与es索引相同的kafka主题？ add_field => {"[@metadata][index]" => "[kafka][topic]"} 对我不起作用
@hiaclibe 不，您需要在 elasticsearch 输出中使用 index => "%{[kafka][topic]}"