单表同步：//www.greatytc.com/p/2ae44704f887
多表同步：//www.greatytc.com/p/773e4ecfa667

1 修改pipelines.yml

进入docker容器，修改/usr/share/logstash/config/pipelines.yml，新增table2配置

- pipeline.id: table1
  path.config: "/etc/logstash/pipeline/logstash.conf"
- pipeline.id: table2
  path.config: "/etc/logstash/pipeline/update.conf"

2 创建update.conf配置文件

input {
 stdin { }
    jdbc {
        #注意mysql连接地址一定要用ip，不能使用localhost等
        jdbc_connection_string => "jdbc:mysql://172.17.0.3:3306/lmrs"
        jdbc_user => "xxxx"
        jdbc_password => "xxxx"
        #数据库重连尝试
        connection_retry_attempts => "3"
        #数据库连接可用校验超时时间，默认为3600s
        jdbc_validation_timeout => "3600"
        #这个jar包的地址是容器内的地址
        jdbc_driver_library => "/etc/logstash/pipeline/mysql-connector-java-8.0.24.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        #开启分页查询（默认是false）
        jdbc_paging_enabled => "true"
        #单次分页查询条数（默认100000，字段较多的话，可以适当调整这个数值）
        jdbc_page_size => "50000"
        #执行的sql语句
        statement => "SELECT a.id,a.`name`,a.long_name,a.brand_id,a.three_category_id as category_id,a.shop_id,a.price,a.sold_count,a.review_count,a.`status`,a.create_time,a.last_time,b.`name` as category,b.path FROM lmrs_products as a LEFT JOIN lmrs_product_categorys as b ON a.three_category_id = b.id where a.last_time > :sql_last_value"
        #需要记录查询结果某字段的值时，此字段为true，否则默认tracking_colum为timestamp的值
        use_column_value => true
        #是否将字段名转为小写，默认为true（如果具备序列化或者反序列化，建议设置为false）
        lowercase_column_names => false
        #需要记录的字段，同于增量同步，需要是数据库字段
        tracking_column =>last_time
        #记录字段的数据类型
        tracking_column_type => "timestamp"
        #上次数据存放位置
        record_last_run => true
        #上一个sql_last_value的存放路径，必须在文件中指定字段的初始值
        last_run_metadata_path => "/etc/logstash/pipeline/update.txt"
        #是否清除last_run_metadata_path的记录，需要增量同步这个字段的值必须为false
        clean_run => false
        #同步的频率（分 时 天 月 年）默认为每分钟同步一次
        schedule => "* * * * *"
        type => "_doc"
    }
 }

 filter {
   jdbc_streaming {
     jdbc_driver_library => "/etc/logstash/pipeline/mysql-connector-java-8.0.24.jar"
     jdbc_driver_class => "com.mysql.jdbc.Driver"
     jdbc_connection_string => "jdbc:mysql://172.17.0.3:3306/lmrs"
     jdbc_user => "xxxx"
     jdbc_password => "xxxx"
     parameters => {"product_id"=>"id"}
     statement => "select `name`,price from lmrs_product_skus where product_id = :product_id"
     target => "skus"
   }

  jdbc_streaming {
    jdbc_driver_library => "/etc/logstash/pipeline/mysql-connector-java-8.0.24.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_connection_string => "jdbc:mysql://172.17.0.3:3306/lmrs"
    jdbc_user => "dark"
    jdbc_password => "mysql"
    parameters => {"product_id"=>"id"}
    statement => "SELECT c.`name`,f.`name` as `value` FROM (SELECT a.name,a.id FROM lmrs_attributes as a LEFT JOIN lmrs_product_attribute_values as b on a.id = b.attribute_id WHERE b.product_id = :product_id) as c LEFT JOIN(SELECT d.attribute_id,d.name FROM lmrs_attribute_values as d LEFT JOIN lmrs_product_attribute_values as e ON d.id = e.attribute_value_id WHERE product_id = :product_id) as f ON c.id = f.attribute_id GROUP BY f.name"
    target => "attributes"
  }
 }

 output {
     elasticsearch {
        #注意es连接地址一定要用ip，不能使用localhost等
        hosts => "172.17.0.7:9200"
        index => "products"
        document_type => "_doc"
        document_id => "%{id}"
     }
     stdout {
        codec => json_lines
    }
}

注意创建update.txt并给予相关权限。

思路：
主要指定的同步字段为修改时间字段（last_time），通过该字段来判断数据是否更新。
注意数据类型为timestamp。