mysql - 使用 Logstash 增量读取 MySQL

Question

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

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

Answer 1

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

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

tracking_column and 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 .