Logstash jdbc 输入插件从没有唯一 id 或没有时间戳的数据库多次获取数据

Question

我创建了一个名为 mypipeline 的管道，并在 pipeline.yml（多管道）中与其他管道一起执行它。

这里的问题是我既没有任何唯一的 id 或标识符列，也没有显示修改日期的时间戳。

mypipeline.conf

input {
  jdbc {
    id => "database"
    jdbc_driver_library => "my_driver"
    jdbc_driver_class => "my_driver_class"
    jdbc_connection_string => "my_connection"
    jdbc_user => "db_user"
    jdbc_password => "db_password"
    jdbc_paging_enabled => true
    statement_filepath => "mysql_query.sql"
  }
}

filter {
  mutate {
    id => "remove_fields"
    remove_field => [ "@version" ]
  }
}

output {
  elasticsearch {
    id => "els_cluster"
    action => "index"
    index => "my_index"
    template_name => "my_index_template"
    hosts => [
      "https://myhost1.com:9200",
      "https://myhost2.com:9200",
      "https://myhost3.com:9200",
      "https://myhost4.com:9200"
    ]
    cacert => '/mycertificate.cer'
    user => "my_user"
    password => "my_password"

我有两个Logstash节点，数据库中有700万条数据。 如果我在两个节点（或一个节点）上启动logstash，它会继续获取数据并且不会停止。 我使用Logstash 8.11.4版本。

你知道如何用两个节点读取 700 万条数据吗？

Answer 1

据我所知，目前还不可能，你可以参考这个GitHub问题，它已经开放很长时间了。 您可以做的是根据某些字段（例如日期、类别或任何其他字段）分隔数据，并在 where 子句中使用该字段值来定义多个管道。

假设您有一个包含 A、B 和 C 等值的类别字段。然后您可以使用这些值在类别字段中使用 where 子句定义每个管道。如果您有日期字段，则可以创建一个日期范围并为每个日期范围创建一个管道。

还有一个建议：拥有一个基于唯一指纹的唯一ID，这将确保您的索引中不会有重复的数据。您可以参考this答案以获取相同的方法。