一、准备

1、mysql

我这里准备了个数据库mysqlEs，表User

结构如下

添加几条记录

2、创建elasticsearch索引

curl -XPUT 'localhost:9200/user?pretty'

3、准备mysql-connector

下载地址为https://dev.mysql.com/downloads/connector/j/ 下载后解压可以得到connector的jar包 。

我把它放在了"/usr/share/logstash/bin/config-mysql/mysql-connector-java-8.0.13.jar"这个目录下。

二、 新建mysql与es交互的配置文件

cd到logstash目录bin目录下,新建文件夹config-mysql,打开并写入mysql.conf文件内容

cd /usr/share/logstash/bin
sudo mkdir config-mysql 
cd config-mysql 
vim /usr/share/logstash/bin/config-mysql/mysql.conf

input {
stdin{
}
jdbc {
# 连接的数据库地址和哪一个数据库，指定编码格式，禁用SSL协议，设定自动重连
jdbc_connection_string => "jdbc:mysql://localhost:3306/mysqlEs?characterEncoding=UTF-8&useSSL=false&autoReconnect=true"
# 用户名密码
jdbc_user => "root"
jdbc_password => "root"
# jar包的位置
jdbc_driver_library => "/usr/share/logstash/bin/config-mysql/mysql-connector-java-8.0.13.jar"
# mysql的Driver
jdbc_driver_class => "com.mysql.jdbc.Driver"
　　　 jdbc_default_timezone => "Asia/Shanghai"
jdbc_paging_enabled => "true"
jdbc_page_size => "1000"
#statement_filepath => "config-mysql/test02.sql"
statement => "select * from User"
schedule => "* * * * *"
#索引的类型
type => "user"
}
}

output {
elasticsearch {
hosts => "localhost:9200"
# index名
index => "user"
document_type => "usertype"
# 需要关联的数据库中有有一个id字段，对应索引的id号
document_id => "%{id}"
}
stdout {
codec => json_lines
}
}

 注意，目前我们这个output是所有的input都会输出到这个index的。往往这不是我们想要的。

output {
　　#只输出类型为user的input
　　if [type] == "user" {

　　elasticsearch {

　　        hosts => "localhost:9200"
　　       # index名
　　       index => "user"
　　      document_type => "usertype"
　　      # 需要关联的数据库中有有一个id字段，对应索引的id号
　　      document_id => "%{id}"
　　  }
　　  stdout {
　　      codec => json_lines
　　  }
　　}
}

三、测试同步数据

输入

/usr/share/logstash/bin/logstash -f /usr/share/logstash/bin/config-mysql/mysql.conf

等待一小会

可以看到一分钟执行一次增量同步。

ctrl+c退出

查询看看有没有同步过来

curl -XGET 'localhost:9200/user/_search?pretty'

过程中会遇到一些bug,如权限不够的话改变登录用户为root

mysql数据自动后台同步

有心的同学可能发现，Ctrl+C退出后就不再同步了。这是因为我们的conf文件没有放置在logstash的默认目录下，只需要copy下就可以了。

cp /usr/share/logstash/bin/config-mysql/mysql.conf /etc/logstash/conf.d/
service logstash.service restart

再测试下，发现正常同步了。