使用 Logstash 增量读取 MySQL答案

【问题标题】：Incremental MySQL reading using Logstash使用 Logstash 增量读取 MySQL
【发布时间】：2015-07-16 06:07:02
【问题描述】：

我有一个 MySQL 数据库，随着时间的推移向其中添加了数据。我想使用 Logstash 将数据传输到 Elastic 中以生成一些报告。

有没有办法记住 MySQL 最后传输的记录，并在插入新记录后恢复记录过程？

【问题讨论】：

如果我理解正确，您正在寻找用于 logstash 的 MySQL 输入。你已经看过logstash-input-jdbc了吗？它不能满足您的需求吗？
我无法确定上述插件是否是我正在寻找的，因为它没有关于如何使用它的文档。我要寻找的是具有一个重要功能的 Logstash 的 MySQL 输入；我需要它来支持向 MySQL 添加新数据，即使它完成了所有数据的记录。因此，一旦将新数据引入 MySQL，它必须检测它们并将它们记录在输出中。它会这样做吗？

【解决方案1】：

为了增量更新 Elasticsearch 中的数据，我们必须在数据库中有一个列，以便它可以用作参考。而且我们还必须设置一个时间表来一次又一次地运行配置文件。

假设您想将新添加的行发送到 Elasticsearch，并且您有一个 id 字段，该字段在表中自动递增。然后，您可以告诉 Logstash 仅获取 id 值大于先前获取记录的 id 值的记录。为了告诉 Logstash 跟踪一列以及跟踪哪一列，我们必须使用

tracking_column 和 use_column_value

JDBC 部分中的选项。之后配置文件将如下所示。

input {
  jdbc {
  jdbc_driver_library => "mysql-connector-java-5.1.36-bin.jar"
  jdbc_driver_class => "com.mysql.jdbc.Driver"
  jdbc_connection_string => "jdbc:mysql://localhost:3306/database"
  jdbc_user => "DBUSERNAME"
  jdbc_password => "DBPASSWORD"
  statement => "select * from TABLE where id >:sql_last_value"
  tracking_column => id
  use_column_value => true
  #I have scheduled this to run in every second
  schedule => "* * * * * *"
  }
}
output {
  elasticsearch {
    hosts => "localhost:9200"
    index => "index"
    document_type => "type"
    user => "userName"
    password => "password"
  }
  stdout { 
    codec => rubydebug 
  }
}

:sql_last_value

包含先前发送的记录的 id 值。但是使用 id 字段，您只能发送新添加的记录。如果要根据对数据库中记录所做的更改更新先前发送的 Elasticsearch 中的文档，则必须使用 row_update_time 列（或类似用作参考的内容），每次更改时都会更新表中的记录。您可以找到有关该here 的更多信息。

【讨论】：

【解决方案2】：

logstash jdbc{} 输入可以为后续查询维护状态。在their documentation 中给出了一些示例，但总体思路是您可以在WHERE 子句中使用sql_last_value：

statement => "SELECT id, mycolumn1, mycolumn2 FROM my_table WHERE id > :sql_last_value"

并告诉logstash使用哪个字段来更新sql_last_value：

tracking_column => id

如果你使用这样的id列，还需要设置

use_column_value => true

如果您使用的是日期时间字段，则 use_column_value 应保持默认值 false。

【讨论】：