安裝

tar -zxvf logstash-6.2.2.tar.gz
mv logstash-6.2.2    /home/elk/logstash-6.2.2-01
cp -r  /home/elk/logstash-6.2.2-01   /home/elk/logstash-6.2.2-02

配置

logstash-pipeline01.conf

input{
	kafka{
   			bootstrap_servers => ["172.30.1.199:9092"]
    		client_id => "nic01"
    		group_id => "nic"
    		#從最新的偏移量開始消費
			auto_offset_reset => "latest" 
    		consumer_threads => 7
			#此屬性會將當前topic、offset、group、partition等資訊也帶到message中
    		decorate_events => true
			codec => "json" 
			#陣列型別，可配置多個topic
    		topics => ["test-business","test-peccancy","test-carwash","test-oilcard","test-personalcenter","test-system","test-task","test-consumer"]
			#所有外掛通用屬性,尤其在input裡面配置多個數據源時很有用
			type => "chengniu" 
}
}
output {
			elasticsearch {
			#ES叢集
			hosts => ["172.30.1.1:9250", 	"172.30.1.1:9251","172.30.1.1:9252","172.30.1.1:9253","172.30.1.1:9254"]
			index => "nic-%{[fields][service]}-%{+YYYY.MM.dd}"
}
}
 

logstash-pipeline02.conf

配置同上

啟動

進入bin目錄下執行:

./logstash -f ../config/logstash-pipeline01.conf
或
nohup  ./logstash -f ../config/logstash-pipeline01.conf  &