2

数据库

给定以下PostgreSQL表test（省略了一些列，例如data在管道中使用的列）：

id (uuid)                            | updated_at (timestamp with time zone)
652d88d3-e978-48b1-bd0f-b8188054a920 | 2018-08-08 11:02:00.000000
50cf7942-cd18-4730-a65e-fc06f11cfd1d | 2018-08-07 15:30:00.000000

日志存储

给定具有以下管道（jdbc_*省略）的 Logstash 6.3.2（通过 Docker）：

input {
    jdbc {
        statement => "SELECT id, data, updated_at FROM test WHERE updated_at > :sql_last_value"
        schedule => "* * * * *"
        use_column_value => true
        tracking_column => "updated_at"
        tracking_column_type => "timestamp"
    }
}

filter {
    mutate { remove_field => "updated_at" }
}

output {
    elasticsearch {
        hosts => "elasticsearch:9200"
        index => "test"
        document_id => "%{id}"
    }
}

问题

当此管道第一次运行（或使用clean_run => true）时，我希望它处理两个数据库行（因为sql_last_valueis 1970-01-01 00:00:00.000000）并将存储的跟踪列的值设置.logstash_jdbc_last_run为2018-08-08 11:02:00.000000000 Z（=所有找到的时间戳中的最新updated_at时间戳）。它将被设置为2018-08-07 15:30:00.000000000 Z，这是两个给定时间戳中较早的一个。这意味着在第二次运行中，将再次处理两行中的另一行，即使它没有更改。

这是预期的行为吗？我是否错过了控制这方面的其他一些配置？

编辑

似乎将使用返回updated_at的最后一行的 （只是尝试了更多行）。所以我必须添加一个ORDER BY updated_at ASC我认为在数据库查询性能方面不是很好的。

日志等

sh-4.2$ cat .logstash_jdbc_last_run
cat: .logstash_jdbc_last_run: No such file or directory

[2018-08-09T14:38:01,540][INFO ][logstash.inputs.jdbc     ] (0.001254s) SELECT id, data, updated_at FROM test WHERE updated_at > '1970-01-01 00:00:00.000000+0000'
sh-4.2$ cat .logstash_jdbc_last_run
--- 2018-08-07 15:30:00.000000000 Z


[2018-08-09T14:39:00,335][INFO ][logstash.inputs.jdbc     ] (0.001143s) SELECT id, data, updated_at FROM test WHERE updated_at > '2018-08-07 15:30:00.000000+0000'
sh-4.2$ cat .logstash_jdbc_last_run
--- 2018-08-08 11:02:00.000000000 Z


[2018-08-09T14:40:00,104][INFO ][logstash.inputs.jdbc     ] (0.000734s) SELECT id, data, updated_at FROM test WHERE updated_at > '2018-08-08 11:02:00.000000+0000'
sh-4.2$ cat .logstash_jdbc_last_run
--- 2018-08-08 11:02:00.000000000 Z
4

1 回答 1

2

从上个月使用 MySQL 到 ES，我遇到了同样的问题。但最终它被解决了。默认情况下，文件 .logstash_jdbc_last_run 会在您的主目录中创建。您可以通过设置 last_run_metadata_path 配置选项来更改此文件的路径。我正在使用 UTC 日期格式。

第一次 sql_last_value 值为 1970-01-01 00:00:00.000000 。它还在logstash_jdbc_last_run 文件中设置日期，这是MySQL 返回的第一条记录。这就是为什么我使用 update_at DESC 的顺序。以下代码对我有用。

input {
    jdbc {
         jdbc_default_timezone => "UTC"
         statement => "SELECT id, data, DATE_FORMAT(updated_at, '%Y-%m-%d %T') as updated_at, FROM test WHERE updated_at > :sql_last_value order by update_at DESC"
         schedule => "* * * * * *"
         use_column_value => true
         tracking_column => "updated_at"
         tracking_column_type => "timestamp"
         last_run_metadata_path => /home/logstash_track_date/.logstash_user_jdbc_last_run" 
    }
}

filter {
    mutate { remove_field => "updated_at" }
}

output {
    stdout { codec => rubydebug }
    elasticsearch {
        hosts => "elasticsearch:9200"
        index => "test"
        document_id => "%{id}"
    }
}
于 2019-11-13T10:12:43.390 回答