logstash同步MySQL数据

1.配置值说明 https://www.elastic.co/guide/en/logstash/7.5/plugins-inputs-jdbc.html

数据库驱动 mysql-connector-java-5.1.48.jar 放在 logstash-7.5.0/logstash-core/lib/jars 目录下

2.配置logstash-mysql.conf

input {
	jdbc {
		jdbc_driver_library => "mysql-connector-java-5.1.48.jar"
		jdbc_driver_class => "com.mysql.jdbc.Driver"
		jdbc_connection_string => "jdbc:mysql://192.168.124.4:3306/xx"
		jdbc_user => "root"
		jdbc_password => "123456"
		schedule => "* * * * *"
		jdbc_paging_enabled => "true"
		jdbc_page_size => "50000"
		statement => "SELECT * FROM t_device WHERE update_time >= :sql_last_value  and update_time<now() order by update_time desc"
		type => "t_device"
		lowercase_column_names => false
		record_last_run => true
		use_column_value => true
		tracking_column => "update_time"
		tracking_column_type => "timestamp"
		last_run_metadata_path => "/datas/elastic/logstash/position/t_device"
		clean_run => false
	}
	jdbc {
		jdbc_driver_library => "mysql-connector-java-5.1.48.jar"
		jdbc_driver_class => "com.mysql.jdbc.Driver"
		jdbc_connection_string => "jdbc:mysql://192.168.124.4:3306/xx"
		jdbc_user => "root"
		jdbc_password => "123456"
		schedule => "* * * * *"
		jdbc_paging_enabled => "true"
		jdbc_page_size => "50000"
		statement => "SELECT * FROM t_area_detail WHERE update_time >= :sql_last_value  and update_time<now() order by update_time desc"
		type => "t_area_detail"
		lowercase_column_names => false
		record_last_run => true
		use_column_value => true
		tracking_column => "update_time"
		tracking_column_type => "timestamp"
		last_run_metadata_path => "/datas/elastic/logstash/position/t_area_detail"
		clean_run => false
	}
}

filter {
	date {
		match => [ "timeslice", "yyyy-MM-dd HH:mm:ss" ]
		timezone => "Asia/Shanghai"
	}
	json {
		source => "message"
		remove_field => ["message"]
	}
}

	output {
	if [type]=="t_device"{
		elasticsearch {
			hosts => ["192.168.33.88:9200"]
			index => "device"
			document_id => "%{id}"
		}
	}
	if [type]=="t_area_detail"{
		elasticsearch {
			hosts => ["192.168.33.88:9200"]
			index => "t_area_detail"
			document_id => "%{id}"
		}
	}
	
	stdout {
		codec => json_lines
		}
}

3.启动 nohuo ./bin/logstash -f config/logstash-mysql.conf &

4.查询数据是否同步