elasticsearch - 在Logstash中为sql_last_value使用表的ID？

Question

我在jdbc输入的logstash插件中有这样的MySQL语句。

statement => "SELECT * from TEST where id > :sql_last_value"

我的表没有这样的 date或 datetime字段。因此，我尝试通过使用 scheduler每分钟检查一次来更新索引，是否在表中添加了任何新行。

我应该只能更新新记录，而不能更新现有记录中的现有值更改。因此，要做到这一点，我需要输入 logstash:

input {
    jdbc {
        jdbc_connection_string => "jdbc:mysql://myhostmachine:3306/mydb" 
        jdbc_user => "root"
        jdbc_password => "root"
        jdbc_validate_connection => true
        jdbc_driver_library => "/mypath/mysql-connector-java-5.1.39-bin.jar"
        jdbc_driver_class => "com.mysql.jdbc.Driver"
        jdbc_paging_enabled => "true"
        jdbc_page_size => "50000"
        schedule => "* * * * *"
        statement => "SELECT * from mytable where id > :sql_last_value"
        use_column_value => true
        tracking_column => id
        last_run_metadata_path => "/path/.logstash_jdbc_last_run"
        clean_run => true
    }
}

因此，每当我创建索引并运行此 logstash文件以上传文档时，它根本就不会上传。文档数显示为零。我确保在运行 .logstash_jdbc_last_run conf文件之前删除了 logstash。

logstash控制台输出的一部分:

[2016-11-02T16:33:00,294][INFO ][logstash.inputs.jdbc ] (0.002000s) SELECT count(*) AS count FROM (SELECT * from TEST where id > '2016-11-02 11:02:00') AS t1 LIMIT 1

并且通过逐分钟检查是正确的继续进行，但是随后它没有得到记录。它是如何工作的？

我想念什么吗？任何帮助，将不胜感激。

Answer 1

您需要像这样修改logstash配置:

jdbc { 
  jdbc_connection_string => "jdbc:mysql://myhostmachine:3306/mydb" 
  jdbc_user => "root" 
  jdbc_password => "root" 
  jdbc_validate_connection => true 
  jdbc_driver_library => "/mypath/mysql-connector-java-5.1.39-bin.jar" 
  jdbc_driver_class => "com.mysql.jdbc.Driver" 
  jdbc_paging_enabled => "true" 
  jdbc_page_size => "50000" 
  schedule => "* * * * *" 
  statement => "SELECT * from TEST where id > :sql_last_value" 
  use_column_value => true 
  tracking_column => "id" 
  tracking_column_type => "numeric" 
  clean_run => true 
  last_run_metadata_path => "/mypath/.logstash_jdbc_last_run" 
}

在您的情况下，最后五个设置很重要。即使 .logstash_jdbc_last_run删除了 clean_run => true文件，也要确保删除它。