單表同步：http://www.reibang.com/p/2ae44704f887
多表同步：http://www.reibang.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 創(chuàng)建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"
        #數(shù)據(jù)庫重連嘗試
        connection_retry_attempts => "3"
        #數(shù)據(jù)庫連接可用校驗超時時間蔗草，默認為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"
        #單次分頁查詢條數(shù)（默認100000，字段較多的話枕扫，可以適當調整這個數(shù)值）
        jdbc_page_size => "50000"
        #執(zhí)行的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
        #需要記錄的字段咕宿，同于增量同步，需要是數(shù)據(jù)庫字段
        tracking_column =>last_time
        #記錄字段的數(shù)據(jù)類型
        tracking_column_type => "timestamp"
        #上次數(shù)據(jù)存放位置
        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
    }
}

注意創(chuàng)建update.txt并給予相關權限芽突。

思路：
主要指定的同步字段為修改時間字段（last_time）试浙，通過該字段來判斷數(shù)據(jù)是否更新。
注意數(shù)據(jù)類型為timestamp寞蚌。