采用logstash-input-jdbc插件的方式
1.安装elasticsearch 采用docker的方式，发现es-head插件官网支持到es5 所以这里安装es5的版本。注意这里只是测试我没有映射docker数据目录跟配置文件。9200端口是http浏览器端口，9300是集群内交互端口
docker pull elasticsearch:5.6.8
docker run -d -p 9200:9200 -p 9300:9300 --name es docker.io/elasticsearch:5.6.8
2.安装elasticsearch-head插件，在浏览器中可以查看ElasticSearch集群并与其进行交互的一款插件，官网地址https://github.com/mobz/elasticsearch-head，这里也是用docker方式部署
docker run -d -p 9100:9100 --name es-manager mobz/elasticsearch-head:5
浏览器访问http://ip:9100
image.png
如果发现点连接 连接不上，则需要修改下es的配置文件，由于es的配置并没有映射出来，所以我们进入到容器中修改，/usr/share/elasticsearch/config/elasticsearch.yml
docker exec -it es /bin/bash
vim /usr/share/elasticsearch/config/elasticsearch.yml
在文件中添加以下内容，重启重启，这样就可以连上了
http.cors.enabled: true
http.cors.allow-origin: "*"
http.cors.allow-headers: Authorization
image.png
3.安装logstash 然后安装logstash-input-jdbc插件，等待安装完成，可能时间会比较慢。
cd /home
wget https://artifacts.elastic.co/downloads/logstash/logstash-5.6.16.tar.gz
tar -xzf logstash-5.6.16.tar.gz
cd logstash-5.6.16/
./bin/logstash-plugin install logstash-input-jdbc
4.下载工具包，设置mysql相关配置 配置文件 mysql.conf
cd /home/logstash-5.6.16/bin/
mkdir config-mysql
cd config-mysql/
wget https://dev.mysql.com/get/Downloads/Connector-J/mysql-connector-java-8.0.13.tar.gz
tar -zxf mysql-connector-java-8.0.13.tar.gz
vim mysql.conf
input {
jdbc {
# mysql相关jdbc配置
jdbc_connection_string => "jdbc:mysql://ip:3306/test?useUnicode=true&characterEncoding=utf-8&useSSL=false"
jdbc_user => "root"
jdbc_password => "123456"
# jdbc连接mysql驱动的文件目录，可去官网下载:https://dev.mysql.com/downloads/connector/j/
jdbc_driver_library => "/home/logstash-5.6.16/bin/config-mysql/mysql-connector-java-8.0.13/mysql-connector-java-8.0.13.jar"
# the name of the driver class for mysql
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_paging_enabled => true
jdbc_page_size => "50000"
jdbc_default_timezone =>"Asia/Shanghai"
# mysql文件, 也可以直接写SQL语句在此处，如下：
statement => "select * from book_info where update_time >= :sql_last_value"
#statement_filepath => "./config/jdbc.sql"
# 这里类似crontab,可以定制定时操作，比如每分钟执行一次同步(分 时 天 月 年)
schedule => "* * * * *"
#type => "jdbc"
# 是否记录上次执行结果, 如果为真,将会把上次执行到的 tracking_column 字段的值记录下来,保存到 last_run_metadata_path 指定的文件中
#record_last_run => true
# 是否需要记录某个column 的值,如果record_last_run为真,可以自定义我们需要 track 的 column 名称，此时该参数就要为 true. 否则默认 track 的是 timestamp 的值.
use_column_value => true
# 如果 use_column_value 为真,需配置此参数. track 的数据库 column 名,该 column 必须是递增的. 一般是mysql主键
tracking_column => "update_time"
tracking_column_type => "timestamp"
last_run_metadata_path => "./logstash_capital_bill_last_id"
# 是否清除 last_run_metadata_path 的记录,如果为真那么每次都相当于从头开始查询所有的数据库记录
clean_run => false
#是否将 字段(column) 名称转小写
lowercase_column_names => false
}
}
output {
elasticsearch {
hosts => "esip:9200"
index => "mysql_order"
document_id => "%{id}"
template_overwrite => true
}
# 这里输出调试，正式运行时可以注释掉
stdout {
codec => json_lines
}
}
注意：sql_last_value这个值比较重要
标志目前logstash同步的位置信息（类似offset）。比如id、updatetime。logstash通过这个标志，可以判断目前同步到哪一条数据。
5.在logstash bin目录下执行，会把现有的数据同步到es中，数据库中新增一条数据也会对应同步过去，注意这里是根据数据库字段update_time 做 增量同步
./logstash -f config-mysql/mysql.conf
image.png
6.完成，在head插件中可以看到es的数据
