1、准备：
1) 启动前面搭建的ES集群， 192.168.127.130， 192.168.127.128， 192.168.127.129
2) 准备要同步的数据库和数据 数据库所在的服务器IP为192.168.1.104 数据库端口为3306 数据库名shop，表名items
items的表结构如下
items表的数据如下
这里有两条数据，实际环境会有更多的数据。
3) 创建items索引
5个分片和1个副本
2、版本信息
ElasticSearch 7.4.2
Logstash 7.4.2
3、Logstash介绍
数据采集，从数据库中采集数据到ElasticSearch
以id或者update_time 作为同步边界
logstash-input-jdbc插件。
预先创建索引
注意Logstash的版本要与ES的版本保持一致。
4、Logstash下载
官网下载太慢，建议下载地址， https://www.newbe.pro/Mirrors/Mirrors-Logstash/
我这里下载的版本为7.4.2
登录192.168.127.130 服务器，下面的操作也是在130这台服务器中进行。
cd /home/tools
tar -zxvf logstash-7.4.2.tar.gz
mv logstash-7.4.2 /usr/local/
安装JDK，我这里安装的是1.8版本
5、同步配置
cd /usr/local/logstash-7.4.2
mkdir sync
1) items.sql
cd sync
vi items.sql
2) 将mysql-connector-java-5.1.48.jar 拷贝到/usr/local/logstash-7.4.2/logstash-core/lib/jars
3) 配置文件
cd /usr/local/logstash-7.4.2/sync
vi logstash-db-sync.conf
input {
jdbc {
# 设置 MySql/MariaDB 数据库url以及数据库名称
jdbc_connection_string => "jdbc:mysql://192.168.1.104:3306/shop?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true"
# 用户名和密码
jdbc_user => "root"
jdbc_password => "123456"
# 数据库驱动所在位置，可以是绝对路径或者相对路径
jdbc_driver_library => "/usr/local/logstash-7.4.2/logstash-core/lib/jars/mysql-connector-java-5.1.48.jar"
# 驱动类名
jdbc_driver_class => "com.mysql.jdbc.Driver"
# 开启分页
jdbc_paging_enabled => "true"
# 分页每页数量，可以自定义
jdbc_page_size => "1000"
# 执行的sql文件路径
statement_filepath => "/usr/local/logstash-7.4.2/sync/items.sql"
# 设置定时任务间隔 含义：分、时、天、月、年，全部为*默认含义为每分钟跑一次任务
schedule => "* * * * *"
# 索引类型
type => "_doc"
# 是否开启记录上次追踪的结果，也就是上次更新的时间，这个会记录到 last_run_metadata_path 的文件
use_column_value => true
# 记录上一次追踪的结果值
last_run_metadata_path => "/usr/local/logstash-7.4.2/sync/track_time"
# 如果 use_column_value 为true， 配置本参数，追踪的 column 名，可以是自增id或者时间
tracking_column => "update_time"
# tracking_column 对应字段的类型
tracking_column_type => "timestamp"
# 是否清除 last_run_metadata_path 的记录，true则每次都从头开始查询所有的数据库记录
clean_run => false
# 数据库字段名称大写转小写
lowercase_column_names => false
}
}
output {
elasticsearch {
# es地址
hosts => ["192.168.127.130:9200","192.168.127.128:9200","192.168.127.129:9200"]
# 同步的索引名
index => "items"
# 设置_docID和数据相同
document_id => "%{id}"
# document_id => "%{itemId}"
}
# 日志输出
stdout {
codec => json_lines
}
}
补充： 如果 192.168.1.130 连接不上mysql（192.168.1.104），在mysql所在的服务器中修改访问权限
如允许所有的IP访问Mysql
GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' IDENTIFIED BY '123456';
6、启动同步
cd /usr/local/logstash-7.4.2/bin
./logstash -f /usr/local/logstash-7.4.2/sync/logstash-db-sync
如上图所示，数据已经在同步了。
7、查看同步的数据
可以看到，数据已经从MySQL同步到了ES中了。