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以上的版本，必定要设置多个索引