Logstash 读取相同的数据（重复）答案

【问题标题】：Logstash Reading Same Data (Duplicates)Logstash 读取相同的数据（重复）
【发布时间】：2022-08-23 18:41:13
【问题描述】：

我正在使用logstash 输入jdbc 插件来读取一个数据库并将数据发送到elasticsearch。我的 Logstash.conf 文件如下所示：

input {
jdbc {
    jdbc_driver_library => \"${LOGSTASH_JDBC_DRIVER_JAR_LOCATION}\"
    jdbc_driver_class => \"${LOGSTASH_JDBC_DRIVER}\"
    jdbc_connection_string => \"${LOGSTASH_JDBC_URL}\"
    jdbc_user => \"${LOGSTASH_JDBC_USERNAME}\"
    jdbc_password => \"${LOGSTASH_JDBC_PASSWORD}\"
    schedule => \"* * * * *\"
    statement => \"select * from testtable\"
    use_column_value => true
    tracking_column => \"time\"
}
}

filter {
  mutate {
    add_field => { \"message\" => \"%{time}\" }
          convert => [ \"time\", \"string\" ]
  }
  date {
      timezone => \"Etc/GMT+3\"
      match => [\"time\" , \"ISO8601\", \"yyyy-MM-dd HH:mm:ss.SSS\"]
      target => \"@timestamp\"
      remove_field => [ \"time\", \"timestamp\" ]
  }
  fingerprint {
    source => [\"testid\", \"programid\", \"unitid\"]
    target => \"[@metadata][fingerprint]\"
    method => \"MD5\"
    key => \"${LOGSTASH_JDBC_PASSWORD}\"
  }
  ruby {
    code => \"event.set(\'[@metadata][tsprefix]\', event.get(\'@timestamp\').to_i.to_s(16))\"
  }
}

output {
    elasticsearch {
        hosts => [\"${LOGSTASH_ELASTICSEARCH_HOST}\"]
        user => \"${ELASTIC_USER}\"
        password => \"${ELASTIC_PASSWORD}\"
        index => \"test\"
        document_id => \"%{[@metadata][tsprefix]}%{[@metadata][fingerprint]}\"
    }
    stdout { codec => json_lines }
}

我尝试在没有这些行的情况下使用这个 .conf：

    use_column_value => true
    tracking_column => \"time\"

还尝试使用：

    clean_run => true

但是 Logstash 会一遍又一遍地读取相同的数据。你能帮我理解为什么 Logstash 一直在读吗？ Logstash (8.3.1) 数据库 (PostgreSQL 14.5) JDBC (42.4.1)

标签： postgresql jdbc logstash

【解决方案1】：

您的 jdbc 输入配置中的语句查询 "select * from testtable" 将在每次运行时从 DB 表中读取所有内容。输入配置应如下所示，以避免重复读取相同的数据。

jdbc {
    jdbc_driver_library => "${LOGSTASH_JDBC_DRIVER_JAR_LOCATION}"
    jdbc_driver_class => "${LOGSTASH_JDBC_DRIVER}"
    jdbc_connection_string => "${LOGSTASH_JDBC_URL}"
    jdbc_user => "${LOGSTASH_JDBC_USERNAME}"
    jdbc_password => "${LOGSTASH_JDBC_PASSWORD}"
    schedule => "* * * * *"
    statement => "select * from testtable where time > :sql_lat_value"
    use_column_value => true
    tracking_column => "time"
    record_last_run => true
    last_run_metadata_path => <valid file path>
}

【讨论】：