近实时数据同步: 使用logstash将数据从多个mysql同步到一个mysql库

1 将数据从mysql到mysql需要两个插件logstash-input-jdbc和logstash-output-jdbc,自行搜索安装

2 多个库同步到一个库,要考虑主键问题.测试时候,源表使用了主键自增长,导致到目标库的数据会冲突,测试的时候,重新生成主键,避免目标库主键不冲突.

3 脚本是数据增量同步脚本.源表中要有update_time字段,来支持数据增量同步.

4 源表的update_time字段是date_time类型,使用logstash同步到目标表的时候,有时区问题.解决方案,是在sql中用date_format函数将数据转化成字符串, 到目标库的时候,将字符串插入.则不会存在时区问题.

5 logstash做数据增量同步的时候,只支持numeric和timestamp类型,下面的测试脚本是把datetime字段拿来当作timestamp来用了

6 没有考虑目标库表中有索引信息的情况

mysql库使用的是mysq8以上版本
logstash-7.3.0版本

源库数据建表脚本

CREATE TABLE `tb_test_logstash` (
  `id` int NOT NULL AUTO_INCREMENT,
  `message` varchar(100) DEFAULT NULL,
  `code` varchar(10) DEFAULT NULL,
  `network_type` int DEFAULT NULL,
  `update_time` datetime DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=101 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci

目标库建表脚本

CREATE TABLE `tb_test_logstash` (
  `meger_unique_id` varchar(255) NOT NULL,
  `id` int(11) DEFAULT NULL,
  `message` varchar(100) DEFAULT NULL,
  `code` varchar(10) DEFAULT NULL,
  `network_type` int(1) DEFAULT NULL,
  `update_time` datetime DEFAULT NULL,
  PRIMARY KEY (`meger_unique_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci

目标表比源表多了一个meger_unique_id字段,是为了解决多个源表导入目标库时候,主键冲突问题.而meger_unique_id字段是用源表的code + network_type + id字段生成,作为目标表的主键.在下面的同步脚本可以看到.

下面数据同步脚本,如需要测试,请修改路径信息和数据库密码以及配置信息

input {
  stdin {
  }
  jdbc {
    jdbc_connection_string => "jdbc:mysql://192.168.1.15:3308/datacenter"
    jdbc_user => "root"
    jdbc_password => "******"
    jdbc_driver_library => "/home/fyk/tool/logstash-7.3.0/bin/config-mysql/mysql-connector-java-8.0.16.jar"
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    jdbc_paging_enabled => "true"
    statement => "select meger_unique_id, id, message, code, network_type, date_format(update_time, '%Y-%m-%d %H:%i:%s') update_time from (select concat(code, network_type, id) as meger_unique_id, id, message, code, network_type,  
update_time from tb_test_logstash ) t where t.update_time >= :sql_last_value"
    record_last_run => true
    last_run_metadata_path => "/home/fyk/tool/logstash-7.3.0/bin/config-mysql/logid/192.168.1.15_last_id"
    use_column_value => true
    tracking_column => "update_time"
    tracking_column_type => "timestamp"
    schedule => "* * * * *"
  }
  jdbc {
    jdbc_connection_string => "jdbc:mysql://192.168.94.128:3306/test_logstash"
    jdbc_user => "root" 
    jdbc_password => "******"
    jdbc_driver_library => "/home/fyk/tool/logstash-7.3.0/bin/config-mysql/mysql-connector-java-8.0.16.jar"
    jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
    jdbc_paging_enabled => "true"
    statement => "select meger_unique_id, id, message, code, network_type, date_format(update_time, '%Y-%m-%d %H:%i:%s') update_time from (select concat(code, network_type, id) as meger_unique_id, id, message, code, network_type, u
pdate_time from tb_test_logstash ) t where t.update_time >= :sql_last_value"
    record_last_run => true
    last_run_metadata_path => "/home/fyk/tool/logstash-7.3.0/bin/config-mysql/logid/192.168.94.128_last_id"
    use_column_value => true
    tracking_column => "update_time"
    tracking_column_type => "timestamp"
    schedule => "* * * * *"
  }
}

output {
  jdbc {
    driver_jar_path=>"/home/fyk/tool/logstash-7.3.0/bin/config-mysql/mysql-connector-java-8.0.16.jar"
    driver_class => "com.mysql.cj.jdbc.Driver"
    connection_string => "jdbc:mysql://127.0.0.1:3306/test_ana?user=root&password=******"
    statement => ["insert into tb_test_logstash (meger_unique_id, id, message, code, network_type, update_time) values (?, ?, ?, ?, ?, ?)  on duplicate key update id = values(id), message = values(message), code = values(code), net
work_type = values(network_type), update_time = values(update_time)", "meger_unique_id", "id", "message", "code", "network_type", "update_time" ]
  }
  stdout {
    codec => json_lines
  }
}

最后编辑于：2020.06.08 08:27:39