如何在Logstash的JDBC输入配置中格式化sql_last_value？

问题根源 #

数据库中modified_dt字段为秒级精度（如2024-05-20 12:34:56），而Logstash默认记录的:sql_last_value为毫秒级（如2024-05-20 12:34:56.789）。直接用>比较时，数据库内同秒的数据会因时间精度差异被遗漏。

解决方案：可以对:sql_last_value进行格式化，两种常用方式： #

1. 在SQL语句中通过数据库函数转换精度

根据你使用的数据库类型，用对应函数将毫秒级的:sql_last_value转成秒级格式，再与modified_dt比较：

• MySQL示例：

  statement => "SELECT * FROM company_table WHERE modified_dt > DATE_FORMAT(:sql_last_value, '%Y-%m-%d %H:%i:%s')"
  

  或用CAST转换：

  statement => "SELECT * FROM company_table WHERE modified_dt > CAST(:sql_last_value AS DATETIME)"
  

• PostgreSQL示例：

  statement => "SELECT * FROM company_table WHERE modified_dt > TO_CHAR(:sql_last_value, 'YYYY-MM-DD HH24:MI:SS')::TIMESTAMP"
  

• Oracle示例：

  statement => "SELECT * FROM company_table WHERE modified_dt > TRUNC(:sql_last_value, 'SS')"
  

2. 配置Logstash跟踪数据库原生字段

通过use_column_value和tracking_column参数，让Logstash直接存储数据库modified_dt的秒级值作为:sql_last_value，从根源避免精度差异：

input {
  jdbc {
    # JDBC驱动及连接配置
    statement => "SELECT * FROM company_table WHERE modified_dt > :sql_last_value"
    last_run_metadata_path => "c/es/timestamp_company.txt"
    use_column_value => true
    tracking_column => "modified_dt"
    tracking_column_type => "timestamp"
  }
}

注意事项 #

如果已存在timestamp_company.txt文件，建议先清空或修改其中的时间值为数据库秒级格式（如2024-05-20 12:34:56），避免首次查询时因格式不匹配出错。

内容的提问来源于stack exchange，提问作者Tejeshvi Rane