利用logstash同步数据至elasticsearch

1.安装

需要安装的应用：logstash，elasticsearch（这两项安装参考我的博客相关介绍），mysql连接器：mysql-connector-java-5.1.34.jar，下载地址：https://dev.mysql.com/downloads/connector/j/5.1.html

2.配置

（1）进入logstash创建目录，我这里在bin目录下创建了mysql目录

将刚刚下载的MySQL连接器的文件解压，将mysql-connector-java-5.1.48.jar拷贝到你刚刚创建的mysql目录下（我这里是/usr/local/logstash-6.4.2/bin/mysql）：


（2）编辑导入配置文件，同样在刚刚创建的mysql目录下创建jdbc.conf文件（此文件将配置数据库连接信息、查询数据sql、分页信息、同步频率等核心信息。）

input {
  stdin {}
  jdbc {
	# Mysql依赖包路径
    jdbc_driver_library => "D:\logstash-6.4.2\bin\mysql\mysql-connector-java-5.1.48.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
	#数据库连接地址
    jdbc_connection_string => "jdbc:mysql://数据库ip:port/表名?characterEncoding=UTF-8&autoReconnect=true"
	# 数据库连接账号密码
    jdbc_user => ""
    jdbc_password => ""
	# 数据库连接尝试数
	connection_retry_attempts => "3"
	# 判断数据库连接是否可用，默认false不开启
	jdbc_validate_connection => "true"
	# 数据库连接可用校验超时时间，默认3600S
	jdbc_validation_timeout => "3600"
	# 开启分页查询，默认false不开启
    jdbc_paging_enabled => "true"
	# 单次分页查询条数（默认100000,若字段较多且更新频率较高，建议调低此值）
    jdbc_page_size => "1000"
	# sql查询语句，sql语句复杂的时候可以通过statement_filepath字段指定sql语句存放的文件路径
    statement => 'select * from fb_ads'
	# 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）；
	lowercase_column_names => false
	# 设置sql查询日志等级
	sql_log_level => warn
	# 是否记录上次执行结果，true表示会将上次执行结果的tracking_column字段的值保存到last_run_metadata_path指定的文件中；
	record_last_run => true
	# 需要记录查询结果某字段的值时，此字段为true，否则默认tracking_column为timestamp的值；
	use_column_value => true
	# 需要记录的字段，用于增量同步，需是数据库字段
	tracking_column => "spider_date"
	# 记录追踪字段类型，ModifyTime时间戳形式，numeric为整数类型
	tracking_column_type => numeric
	# record_last_run上次数据存放位置；
	last_run_metadata_path => "mysql/last_id.txt"
	# 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false；
	clean_run => false
	# 同步频率(分 时 天 月 年)，默认每分钟同步一次；
	schedule => "* * * * *"
  }
}

output {
    elasticsearch {
		# 索引名字，必须小写
        index => "shopify"
		# ES集群地址,列["192.168.1.1:9200", "192.168.1.2:9200", "192.168.1.3:9200"]
        hosts => "localhost:9200"
		# 数据唯一索引（建议使用数据库KeyID）
        document_id => "%{id}"
		# 为elasticsearc设置映射关系，template_name为表明，template为mapping配置文件的路径
		template_name => "fb_ads"
		template_overwrite => true
		template => "/usr/local/logstash-6.4.2/template/fb_ads.json"
    }
    stdout {
        codec => json_lines
    }
}

（3）上面讲的配置文件适合单张表导入，如果我们要同时导入多张表怎么办，那么今天他来了

input {
  stdin {}
  jdbc {
	 type => "fb_ads"
   	 jdbc_driver_library => "/usr/local/logstash-6.4.2/bin/mysql/mysql-connector-java-5.1.48.jar"
   	 jdbc_driver_class => "com.mysql.jdbc.Driver"
   	 jdbc_connection_string => "jdbc:mysql://数据库ip:port/表名?characterEncoding=UTF-8&autoReconnect=true"
   	 jdbc_user => ""
   	 jdbc_password => ""
	 connection_retry_attempts => "3"
	 jdbc_validate_connection => "true"
	 jdbc_validation_timeout => "3600"
   	 jdbc_paging_enabled => "true"
   	 jdbc_page_size => "500"
	 jdbc_default_timezone =>"Asia/Shanghai"
   	 statement => 'select * from fb_ads where update_time>= :sql_last_value'
	 lowercase_column_names => false
	 sql_log_level => warn
	 record_last_run => true
	 use_column_value => true
	 tracking_column => "update_time"
	 tracking_column_type => numeric
	 last_run_metadata_path => "/usr/local/logstash-6.4.2/bin/mysql/fb_ads_last_id.txt"
	 clean_run => false
	 schedule => "* * * * *"
  }	
    jdbc {
	 type => "fb_homepage"
   	 jdbc_driver_library => "/usr/local/logstash-6.4.2/bin/mysql/mysql-connector-java-5.1.48.jar"
   	 jdbc_driver_class => "com.mysql.jdbc.Driver"
   	 jdbc_connection_string => "jdbc:mysql://数据库ip:port/表命?characterEncoding=UTF-8&autoReconnect=true"
   	 jdbc_user => ""
   	 jdbc_password => ""
	 connection_retry_attempts => "3"
	 jdbc_validate_connection => "true"
	 jdbc_validation_timeout => "3600"
   	 jdbc_paging_enabled => "true"
   	 jdbc_page_size => "500"
	 jdbc_default_timezone =>"Asia/Shanghai"
   	 statement => 'select * from fb_homepage where update_stamp>= :sql_last_value'
	 lowercase_column_names => false
	 sql_log_level => warn
	 record_last_run => true
	 use_column_value => true
	 tracking_column => "update_stamp"
	 tracking_column_type => numeric
	 last_run_metadata_path => "/usr/local/logstash-6.4.2/bin/mysql/fb_homepage_last_id.txt"
	 clean_run => false
	 schedule => "* * * * *"
  }	
}

output {
	if [type] == "fb_ads"{
		elasticsearch {
			index => "fb_ads"
			hosts => "ip:9200"
			document_id => "%{id}"
			template_name => "fb_ads"
			template_overwrite => true
			template => "/usr/local/logstash-6.4.2/template/fb_ads.json"
		}
	}
	if [type] == "fb_homepage"{
		elasticsearch {
			index => "fb_homepage"
			hosts => "ip:9200"
			document_id => "%{id}"
			template_name => "fb_homepage"
			template_overwrite => true
			template => "/usr/local/logstash-6.4.2/template/fb_homepage.json"
		}
	}
    stdout {
        codec => json_lines
    }
}

多张表导入的逻辑与单张表导入差不多，最后在输出的地方加上判断即可。

3.启动

进入logstash的bin目录运行：./logstash -f ./mysql/jdbc.conf，如果想要后台运行的话：nohup ./logstash -f ./mysql/jdbc.conf &

4.性能

（1）如果第一次进行数据导入的时候可能因为数据量较大导致mysql性能下降，这是我们需要将数据按照区间划分进行多次导入。