方法 / 步骤
一:Logstash实现
1.1 安装插件
# 从Logstash的bin目录下安装输入输出ES和MySQL插件
./logstash-plugin install logstash-output-elasticsearch
./logstash-plugin install logstash-input-jdbc
将mysql-connector-java-8.0.11.jar copy到logstash/bin/mysql目录下
1.2 配置
1.2.1 pipelines配置
- logstash主目录下\config\pipelines.yml 放开下面的注释
- pipeline.id: test
pipeline.workers: 1
pipeline.batch.size: 1
config.string: "input { generator {} } filter { sleep { time => 1 } } output { stdout { codec => dots } }"
# - pipeline.id: another_test
queue.type: persisted
path.config: "/tmp/logstash/*.config"1.2.2 JDBC 同步配置
1.2.2.1 全量更新
文件名称: init-mysql2es.conf
input {
stdin {}
jdbc{
type => "classify"
jdbc_default_timezone => "Asia/Shanghai"
# mysql 数据库链接,school-edu为数据库名
jdbc_connection_string => "jdbc:mysql://192.168.11.10:3306/ldd_saas?useUnicode=true&characterEncoding=utf8&serverTimezone=UTC"
# 用户名和密码
jdbc_user => "root"
jdbc_password => "useradmin"
# 驱动
jdbc_driver_library => "D:/foundation-server/dev-pkg/elk/logstash-7.0.0/bin/mysql-connector-java-8.0.19.jar"
# 驱动类名
jdbc_driver_class => "com.mysql.jdbc.Driver"
# 是否开启分页
jdbc_paging_enabled => true
#指定每页显示50000条
jdbc_page_size => "50000"
#是否开启记录上次追踪的结果，也就是上次更新的时间，这个会记录到 last_run_metadata_path 的文件
use_column_value => false
#执行的sql 文件路径+名称
#statement_filepath => "D:/foundation-server/dev-pkg/elk/logstash-7.0.0/bin/mysql/jdbc.sql"
#执行的sql语句
statement => "SELECT `member_id` AS id, `club_id`, `name`, `sex`, `mobile`, `avatar`, `birthday`, `face_img`, `credit_type`, `credit_no`, `wechat_no`, `nick_name`, `stature`, `weight` FROM member_basics"
# 设置监听间隔 各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新
#schedule => "* * * * *"
}
}
output {
elasticsearch {
# ES的IP地址及端口
hosts => ["localhost:9200"]
# 索引名称
index => "member_basics"
#user => "elastic"
#password => "changeme"
#document_type => "_doc"
# 自增ID 需要关联的数据库中有有一个id字段，对应索引的id号
document_id => "%{id}"
}
stdout {
# JSON格式输出
codec => json_lines
}
}1.2.2.2 增量更新
文件名称: sync-mysql2es.conf
#logstash输入配置
input {
#jdbc输入配置，用来指定mysql中需要同步的数据查询SQL及同步周期
jdbc {
jdbc_driver_library => "<path>/mysql-connector-java-8.0.16.jar"
jdbc_driver_class => "com.mysql.jdbc.Driver"
jdbc_connection_string => "jdbc:mysql://<MySQL host>:3306/es_db"
jdbc_user => <my username>
jdbc_password => <my password>
# 是否开启分页
jdbc_paging_enabled => true
# 是否开启记录上次追踪的结果，也就是上次更新的时间，这个会记录到 last_run_metadata_path 的文件
use_column_value => true
# 用来控制增量更新的字段，一般是自增id或者创建、更新时间，注意这里要采用sql语句中select采用的字段别名
tracking_column => "unix_ts_in_secs"
# tracking_column 对应字段的类型
tracking_column_type => "numeric"
# 设置定时任务间隔 含义：分、时、天、月、年，全部为*默认含义为每分钟跑一次任务，这里设置为每5分钟同步一次
schedule => "*/5 * * * * *"
# 同步数据的查询sql语句
statement => "SELECT *, UNIX_TIMESTAMP(modification_time) AS unix_ts_in_secs FROM es_table WHERE (UNIX_TIMESTAMP(modification_time) > :sql_last_value AND modification_time < NOW()) ORDER BY modification_time ASC"
}
}
#logstash输入数据的字段匹配和数据过滤
filter {
mutate {
copy => { "id" => "[@metadata][_id]"}
remove_field => ["id", "@version", "unix_ts_in_secs"]
}
}
#logstash输出配置
output {
# 采用stdout可以将同步数据输出到控制台，主要是调试阶段使用
stdout { codec => "rubydebug"}
# 指定输出到ES的具体索引
elasticsearch {
index => "rdbms_sync_idx"
document_id => "%{[@metadata][_id]}"
}
}1.2.3 运行脚本
# 全量构建 run-init2es.bat
logstash -f init-mysql2es.conf
#增量构建 文件名称: run-sync2es.bat
logstash -f sync-mysql2es.conf- 导入成功
- kibana 查看已经导入成功
参考资料 & 致谢
[1] mysql （全量）数据导入到 elasticsearch
[2] 通过Logstash实现mysql数据定时增量同步到ES
