1、下载Logstash和Elasticsearch,需要版本一致,本次使用版本均为7.8.0版本
[root@zjk data]# ll elasticsearch-7.8.0-linux-x86_64.tar.gz logstash-7.8.0.tar.gz
-rw-r--r--. 1 root root 319112561 12月 28 15:39 elasticsearch-7.8.0-linux-x86_64.tar.gz
-rw-r--r--. 1 root root 167162975 1月 24 22:13 logstash-7.8.0.tar.gz
2、Elasticsearch的安装与启动此处不赘述,可浏览 elasticsearch安装部署(单机和集群)
3、因为Logstash是用JRuby语言开发的,故需要先装JDK,此次使用的是jdk1.8
4、此次迁移的数据库是Tidb 5.7.25-TiDB-v6.3.0(Mysql),所以驱动使用Mysql驱动,自行下载mysql-connector-java-8.0.27.jar,放至 logstash-7.8.0/logstash-core/lib/jars/
5、编写配置文件(重要),此处以单表同步为例
input {
jdbc {
type => "_doc"
# 数据库连接地址
jdbc_connection_string => "jdbc:mysql://ip:port/dbname?characterEncoding=UTF-8&autoReconnect=true"
# 数据库连接账号密码;
jdbc_user => "root"
jdbc_password => "root"
# MySQL依赖包路径;
jdbc_driver_library => "/data/logstash-7.8.0/logstash-core/lib/jars/mysql-connector-java-8.0.27.jar"
# the name of the driver class for mysql 低版本mysql 使用 com.mysql.jdbc.Driver
jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
# 数据库重连尝试次数
connection_retry_attempts => "3"
# 判断数据库连接是否可用,默认false不开启
jdbc_validate_connection => "true"
# 数据库连接可用校验超时时间,默认3600S
jdbc_validation_timeout => "3600"
# 开启分页查询(默认false不开启);
jdbc_paging_enabled => "true"
# 单次分页查询条数(默认100000,若字段较多且更新频率较高,建议调低此值);
jdbc_page_size => "100000"
# statement为查询数据sql,如果sql较复杂,建议配通过statement_filepath配置sql文件的存放路径;
# sql_last_value为内置的变量,存放上次查询结果中最后一条数据tracking_column的值,此处即为ModifyTime;
# statement_filepath => "mysql/jdbc.sql"
statement => "select a.* from (select * from tablename order by id) a where a.id > :sql_last_value"
# 是否将字段名转换为小写,默认true(如果有数据序列化、反序列化需求,建议改为false);
lowercase_column_names => false
# Value can be any of: fatal,error,warn,info,debug,默认info;
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 => "id"
# Value can be any of: numeric,timestamp,Default value is "numeric"
tracking_column_type => "numeric"
# record_last_run上次数据存放位置;
last_run_metadata_path => "/data/logstash-7.8.0/config/track_time"
# 是否清除last_run_metadata_path的记录,需要增量同步时此字段必须为false;
clean_run => false
#
# 同步频率(分 时 天 月 年),默认每分钟同步一次;
#
schedule => "* * * * *"
}
}
output {
elasticsearch {
# host => "localhost"
# port => "9200"
# 配置ES地址,集群配置方式 ["192.168.1.1:9200", "192.168.1.2:9200", "192.168.1.3:9200"]
hosts => ["ip:9200"]
# 索引名字,必须小写,一般以同步的表名为索引名称
index => "index_name"
# 数据唯一索引(建议使用数据库KeyID)
document_id => "%{id}"
}
stdout {
codec => json_lines
}
}
6、启动logstash
./logstash -f ../config/logstash-mysql2es.conf
7、如需同时迁移多张表至Elasticsearch,则配置如下
input {
stdin {}
jdbc {
# 多表同步时,表类型区分,建议命名为“库名_表名”,每个jdbc模块需对应一个type;
type => "TestDB_DetailTab"
# 其他配置此处省略,参考单表配置
# ...
# ...
# record_last_run上次数据存放位置;
last_run_metadata_path => "mysql\last_id.txt"
# 是否清除last_run_metadata_path的记录,需要增量同步时此字段必须为false;
clean_run => false
#
# 同步频率(分 时 天 月 年),默认每分钟同步一次;
schedule => "* * * * *"
}
jdbc {
# 多表同步时,表类型区分,建议命名为“库名_表名”,每个jdbc模块需对应一个type;
type => "TestDB_Tab2"
# 多表同步时,last_run_metadata_path配置的路径应不一致,避免有影响;
# 其他配置此处省略
# ...
# ...
}
}
filter {
json {
source => "message"
remove_field => ["message"]
}
}
output {
# output模块的type需和jdbc模块的type一致
if [type] == "TestDB_DetailTab" {
elasticsearch {
# host => "192.168.1.1"
# port => "9200"
# 配置ES集群地址
hosts => ["192.168.1.1:9200", "192.168.1.2:9200", "192.168.1.3:9200"]
# 索引名字,必须小写
index => "detailtab1"
# 数据唯一索引(建议使用数据库KeyID)
document_id => "%{KeyId}"
}
}
if [type] == "TestDB_Tab2" {
elasticsearch {
# host => "192.168.1.1"
# port => "9200"
# 配置ES集群地址
hosts => ["192.168.1.1:9200", "192.168.1.2:9200", "192.168.1.3:9200"]
# 索引名字,必须小写
index => "detailtab2"
# 数据唯一索引(建议使用数据库KeyID)
document_id => "%{KeyId}"
}
}
stdout {
codec => json_lines
}
}
logstash 特性:
- 无需开发,无需开启myslq binlog等,仅需安装配置logstash即可;
- 凡是SQL可以实现的logstash均可以实现(本就是通过sql查询数据)
- 支持每次全量同步或按照特定字段(如递增ID、修改时间)增量同步;
- 同步频率可控,最快同步频率每分钟一次(如果对实效性要求较高,慎用);
- 不支持被物理删除的数据同步物理删除ES中的数据(可在表设计中增加逻辑删除字段IsDelete标识数据删除)。
- 如使用递增ID作为增量同步依据,则要考虑分布式数据库中,生成id为分段式生成,id可能会变小。