logstash-input-jdbc同時同步多個表

同步一個表，能夠參考個人上一篇 logstash-jdbc-input與mysql數據庫同步

同步多個表的作法，跟一個表相似，惟一不一樣的是 .conf 文件中的配置

 在這裏我加了一個腳本文件jdbc-seckill.sql，是爲了查詢第二個表的數據（其實也能夠不要此文件，而是直接將sql寫在 .conf文件中）

SELECT seckill_id id, name, number, create_time 
FROM seckill 
WHERE number > :number

這裏我增長了一個查詢條件，傳值的方式，可在下面的配置文件中看到有個parameters

 

接下來，咱們直接看配置文件：

input {
    jdbc {
      jdbc_connection_string => "jdbc:mysql://dev.yonyouccs.com:3001/test"
      jdbc_user => "root"
      jdbc_password => "root"
      jdbc_driver_library => "D:/software/logstash-6.2.2/logstash-6.2.2/mysql/mysql-connector-java-5.1.40.jar"
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
      statement_filepath => "D:/software/logstash-6.2.2/logstash-6.2.2/mysql/jdbc.sql"
      schedule => "* * * * *"
      type => "user"
    }
    jdbc {
      jdbc_connection_string => "jdbc:mysql://dev.yonyouccs.com:3001/test"
      jdbc_user => "root"
      jdbc_password => "root"
      jdbc_driver_library => "D:/software/logstash-6.2.2/logstash-6.2.2/mysql/mysql-connector-java-5.1.40.jar"
      jdbc_driver_class => "com.mysql.jdbc.Driver"
      jdbc_paging_enabled => "true"
      jdbc_page_size => "50000"
      parameters => {"number" => "200"}
      statement_filepath => "D:/software/logstash-6.2.2/logstash-6.2.2/mysql/jdbc-seckill.sql"
      schedule => "* * * * *"
      type => "kill"
    }
}

filter {
    json {
        source => "message"
        remove_field => ["message"]
    }
}

output {
    if[type] == "user" {
        elasticsearch {
            hosts => ["localhost:9200"]
            index => "index_user"
            document_id => "%{id}"
        }
    }
    if[type] == "kill" {
        elasticsearch {
            hosts => ["localhost:9200"]
            index => "index_kill"
            document_id => "%{id}"
        }
    }
    stdout {
        codec => json_lines
    }
}

能夠看到在input下，咱們又新增了一個jdbc。在jdbc中，咱們新增了一個屬性type，用來在output中作判斷

在output中，咱們沒有傳document_type，而且還使用了兩個index。這是由於在elasticsearch6.0中，一個索引下只能有一個類型，否則會報錯。這裏咱們可

參考ElasticSearch 6.0新特性介紹

 

一樣的，咱們使用cmd執行命令：logstash -f ../mysql/mysql.conf

執行了兩個sql腳本，其中那個帶where條件的sql參數也傳進去了

咱們再來看elasticsearch-head中的數據：

新增了兩個索引index_kill、index_user。同時，每一個索引下還有一個虛擬索引doc

 

踩過的坑

一、配置文件中，當在input的jdbc下，增長type屬性時，會致使該索引下增長type字段。因此sql查詢出的字段不要用type，若是有，as成其餘的名字，否則的話，這裏判斷會有異常

二、同步多個表，elasticsearch6.0以上的版本，必定要設置多個索引