（三）使用logstash-input-kafka消费信息并根据topic写入不同的es索引

末蓝、 2023-05-31 06:55 8阅读 0赞

# 一、logstash-input-kafka介绍 #

\[[Logstash 参考指南（Kafka输入插件）][Logstash _Kafka]

# 二、举例 #

**分析：利用 decorate\_events 参数，将 kafka的元信息（如：topic，group，offset等）全部输出，根据不同的topic即可输出到不同的es索引。**

input{ 
    	kafka{ 
    		bootstrap_servers => ["55.0.10.19:9092,55.0.10.20:9092,55.0.10.21:9092"]
    		group_id => "kafkaToes5" 
    		auto_offset_reset => "earliest"
    		consumer_threads => 5 
    		decorate_events => true
    		topics => ["a1","a2"]
    		//security_protocol => "SASL_PLAINTEXT" #sasl认证使用
          	//sasl_mechanism => "PLAIN" #sasl认证使用
          	//jaas_path => "/var/vcap/jobs/logstash/bin/kafka.conf" #sasl认证使用-认证填写的密码文件
    	}
    }
    filter{ 
    	if [@metadata][kafka][topic] == "a1" { 
    		json{ 
    			source => "message"
    			skip_on_invalid_json => true
    		}
    		mutate {  
                remove_field => ["@version","@timestamp"]
            }
    	}
    	if [@metadata][kafka][topic] == "a2" { 
    		json{ 
    			source => "message"
    			skip_on_invalid_json => true
    		}
    		mutate {  
                remove_field => ["@version","@timestamp"]
            }
    	}
    }
    output { 
    	stdout{ 
    		codec => json_lines
    	}
    	if [@metadata][kafka][topic] == "a1" { 
    		elasticsearch {  
    			index => "test123"
    			hosts => ["55.0.10.10:9280","55.0.10.11:9280"]
    			user => "admin"
    			password => "123456"
    			document_type=> "doc"
    		}
    	}
    	if [@metadata][kafka][topic] == "a2" { 
    		elasticsearch {  
    			index => "test456"
    			hosts => ["55.0.10.10:9280","55.0.10.11:9280"]
    			user => "admin"
    			password => "123456"
    			document_type=> "doc"
    		}
    	}
    }
    
    补充：
    //认证填写的内容例子
    // cat /var/vcap/jobs/logstash/bin/kafka.conf
    KafkaClient { 
        org.apache.kafka.common.security.plain.PlainLoginModule required
        username="kafkaUser"
        password="kafkaPwd";
    };

参考：

[使用logstash的logstash-input-kafka插件读取kafka中的数据][logstash_logstash-input-kafka_kafka]  
[logstash 重复消费kafka问题][logstash _kafka]

[Logstash _Kafka]: https://segmentfault.com/a/1190000016595992
[logstash_logstash-input-kafka_kafka]: https://blog.csdn.net/lvyuan1234/article/details/78653324
[logstash _kafka]: https://www.jianshu.com/p/35e57b10cb33