利用logstash将mysql多表数据增量同步到es

第一次发送sql请求查询,修改时间参数值是为系统最开始的时间（1970年）,可以查询的
到所有大于1970年的数据,并且会将最后一条数据的update_time时间记录下来,
作为下一次定时查询的条件

如何安装,以及如何运行,这里就不做描述,没有装过的,可以参考我的这篇文章
https://www.jianshu.com/p/f52d9c843bd8

docker search mysql

如何安装docker,不是本文重点,这里不做多描述
docker pull mysql:5.7  (这里选择的是第一个mysql镜像， :5.7选择的5.7版本)
docker pull mysql       # 拉取最新版mysql镜像

docker run -p 3306:3306 --name mysql -e MYSQL_ROOT_PASSWORD=123456 -d mysql:5.7

账号:root
密码:123456

DROP DATABASE IF EXISTS `test`;
CREATE DATABASE `test` DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;

USE test;

#商品表
DROP TABLE IF EXISTS `goods`;
CREATE TABLE `goods` (
  `id` int(11) NOT NULL,
  `name` varchar(255) DEFAULT NULL,
  `update_time` datetime DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

INSERT INTO `goods` VALUES (1, '黎明电脑', '2020-07-01 00:40:19');
INSERT INTO `goods` VALUES (2, '黎明手机', '2020-07-01 00:40:32');

#用户表
DROP TABLE IF EXISTS `user`;
CREATE TABLE `user` (
  `id` int(11) NOT NULL,
  `name` varchar(255) DEFAULT NULL,
  `update_time` datetime DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

INSERT INTO `user` VALUES (1, '黎明1', '2020-07-01 00:40:01');
INSERT INTO `user` VALUES (2, '黎明2', '2020-07-01 00:40:09');

官方地址
https://www.elastic.co/cn/downloads/logstash

国内加速下载网址
https://www.newbe.pro/Mirrors/Mirrors-Logstash/

下载地址
wget https://mirrors.huaweicloud.com/logstash/6.7.2/logstash-6.7.2.zip

下载zip命令解压
yum -y install zip

解压
unzip logstash-6.7.2.zip

https://mvnrepository.com/artifact/mysql/mysql-connector-java

安装会有点慢,大概2分钟左右吧
 bin/logstash-plugin install logstash-input-jdbc
 bin/logstash-plugin install logstash-output-elasticsearch

进入logstash/config目录下,新建 user.conf文件
vim user.conf


添加内容
input {
  jdbc {
    jdbc_driver_library => "/usr/local/software/my/mysql-connector-java-5.1.47.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://192.168.137.11:3306/test"
    jdbc_user => "root"
    jdbc_password => "123456"
    schedule => "* * * * *"
    statement => "SELECT * FROM user WHERE update_time >= :sql_last_value"
    use_column_value => true
    tracking_column_type => "timestamp"
    tracking_column => "update_time"
    last_run_metadata_path => "syncpoint_table"
  }
}


output {
    elasticsearch {
        # ES的IP地址及端口
        hosts => ["192.168.137.11:9200"]
        # 索引名称 可自定义
        index => "user"
        # 需要关联的数据库中有有一个id字段，对应类型中的id
        document_id => "%{id}"
        document_type => "user"
    }
    stdout {
        # JSON格式输出
        codec => json_lines
    }
}

./logstash -f ../config/user.conf

规则如下:
一个表,一个配置
多个表,多个配置
需要同步多少表,就需要加多少配置
当然配置的内容都差不多,改的地方是查询的表名,和es的索引以及类型的名称

vim pipelines.yml

- pipeline.id: table1
  path.config: "/usr/local/software/my/logstash-6.7.2/config/user.conf"
- pipeline.id: table2
  path.config: "/usr/local/software/my/logstash-6.7.2/config/goods.conf"

进入bin目录
./logstash