logstash提取mysql数据库表数据输出到elasticsearch中

下载地址: https://www.elastic.co/cn/downloads/logstash
下载到本地是一个zip压缩文件,解压文件

input{
#标准输入
stdin { }
#配置jdbc插件
 jdbc { 
	#配置jdbc的驱动jar包的位置,这里需要手动下载好mysql-connector-java-8.0.13.jar包,配置的值为存储jar包的绝对路径
    jdbc_driver_library => "D:softwarelogstashmysql-connector-java-8.0.13.jar" 
	#配置jdbc驱动的类
    jdbc_driver_class => "com.mysql.jdbc.Driver" 
	#jdbc连接数据库ip地址数据库名称
    jdbc_connection_string => "jdbc:mysql://192.168.1.124:3306/trackdata20220112"
	#数据库账号
    jdbc_user => "root"
	#数据库密码
	jdbc_password => "root"
	#指定多久执行一次数据输出,把mysql新的数据输出到es中
    schedule => "* * * * *" 
	#追踪数据库表主键字段
	tracking_column => "session_id"
	#追踪字段的类型，目前只有数字(numeric)和时间类型(timestamp)，默认是数字类型
	tracking_column_type => "numeric"
	#值为ture才能开启追踪
	use_column_value => true
	#查询的sql语句,sql_last_value为系统默认参数,表示sql最后执行的主键id值,只有大于这个值的数据才会输出到es中,last_run_metadata_path指定这个值的存储位置
	statement => "select * from session_history where session_id > :sql_last_value"	
	#JDBC 启用分页
	jdbc_paging_enabled => true
	#每页查询多少条
	jdbc_page_size => "500"
	last_run_metadata_path => "D:softwarelogstashjdbc-position.txt"
	#开启记录最后一次运行的结果
	record_last_run => true
 } 
}

filter{
#使用geoio插件把根据ip值生成对应的地理对象
geoip {
				#插件用到的字段
    	        source => "ip"
				#插件名称
    	        target => "geoip"
	}

}



output {
	#配置es
	elasticsearch { 
               #自己的es地址
		hosts => ["http://localhost:9200"] 
		#索引名称,这里注意索引名称需要和es中映射模板index_patterns参数规则匹配,否则ip插件生成的地理对象类型不对
		index => "logstash-session_history" 
		#索引类型
		document_type => "_doc"
	} 
	#标准输出
	stdout {
		codec => json_lines 
	} 
}

存储路径和.conf文件中last_run_metadata_path 属性的值一致,内容logstash会自动帮我们生成,内容就是.conf中追踪数据库表主键字段值,会记录最后一个追踪字段的值,logstash下次输出数据到es就是输出大于这个值的全部数据

logstash.bat -f .conf文件存储的路径

logstash默认使用Unix时间,同步到es中的时间 timestamp 需要加八小时