Logstash输入Kafka输出Es配置

# Logstash配置文件
input {
  kafka {
    bootstrap_servers => "kafka-server1:9092,kafka-server2:9092"
    topics => ["your_topic"]
    group_id => "logstash_group"
    consumer_threads => 3
    codec => "json"
  }
}
 
filter {
  # 在这里添加任何需要的过滤器配置
}
 
output {
  elasticsearch {
    hosts => ["http://es-server1:9200", "http://es-server2:9200"]
    index => "your_index"
    document_type => "your_type"
    document_id => "%{your_id_field}"
  }
}

这个配置文件定义了Logstash的输入、过滤和输出。输入是Kafka，输出是Elasticsearch。在Kafka输入插件中，你需要指定Kafka集群地址、消费的topic、消费组ID和消费者线程数。同时，使用了json编解码器。在Elasticsearch输出插件中，你需要指定Elasticsearch节点地址、索引名称、文档类型和文档ID。这样，从Kafka消费的数据会被索引到Elasticsearch中。