Logstash管道（Elasticsearch输入转SQL输出）重复数据问题咨询

问题背景 #

使用Logstash将Elasticsearch索引数据通过JDBC输出到SQL Server表时，出现重复数据问题，尝试过uuid过滤器但无效。配置如下：

input{
    elasticsearch {
        hosts => "ip:9200"
        index => "indexname"
        user => "user"
        password => "elastic"
        query => '{ "query": { "query_string": { "query": "*" } } }'
        schedule => "*/5 * * * *"   #每5分钟执行一次查询
        size => 1500   #每次查询最多拉取1500条文档
        scroll => "5m" #搜索上下文保持5分钟
        docinfo => true
      }
}
filter {
     uuid {
        target    => "document_id"
        overwrite => true
      }
   }
    
output {
  if "API_REQUEST" in [message] {
    jdbc {
      driver_jar_path => '/usr/share/logstash/vendor/jar/jdbc/mssql-jdbc-12.2.0.jre8.jar'
      connection_string => "jdbc:sqlserver://ip:1433;databaseName=izdb;user=user;password=pass;ssl=false;trustServerCertificate=true"
      enable_event_as_json_keyword => true
      statement => [
"INSERT INTO Transaction (document_id, logLevel, timestamp) VALUES (?,?,?)",
        "document_id",
        "logLevel",
        "timestamp"
      ]
    }
  }
}

重复数据的核心原因 #

• 全量重复拉取：ES输入插件每5分钟执行一次全量查询（query: "*"），每次都会拉取所有符合条件的文档，没有做增量过滤，导致旧数据被反复插入。
• UUID生成无意义：uuid过滤器是每次运行时生成新的随机UUID，并非基于ES文档的唯一标识。同一条ES文档每次被拉取时，都会生成不同的document_id，SQL的INSERT语句没有唯一约束，自然会重复插入。
• 未利用ES自带唯一ID：ES文档本身有唯一的_id字段（已通过docinfo: true加载到[@metadata][_id]），却弃而不用，失去了判断重复的核心依据。

可行解决方案 #

方案1：基于ES文档唯一ID实现幂等性（推荐） #

这是最可靠的方案，同时解决重复拉取和重复插入问题：

1. 修改ES输入配置，实现增量拉取

开启记录上次同步时间，只拉取上次同步后新增/更新的文档：

input{
    elasticsearch {
        hosts => "ip:9200"
        index => "indexname"
        user => "user"
        password => "elastic"
        # 增量查询：只拉取上次同步时间之后的文档
        query => '{ "query": { "range": { "@timestamp": { "gte": "%{[@metadata][last_run]}" } } } }'
        schedule => "*/5 * * * *"
        size => 1500
        scroll => "5m"
        docinfo => true
        # 开启记录上次运行时间
        record_last_run => true
        # 指定存储上次运行时间的文件路径，确保Logstash有读写权限
        last_run_metadata_path => "/var/lib/logstash/.es_last_run"
      }
}

2. 替换UUID过滤器，使用ES自带的_id

filter {
  mutate {
    # 将ES文档的唯一ID赋值给document_id字段
    add_field => { "document_id" => "%{[@metadata][_id]}" }
  }
}

3. 修改SQL语句为MERGE（SQL Server幂等插入语法）

先给Transaction表的document_id字段添加唯一约束，然后用MERGE语句实现"存在则更新，不存在则插入"：

output {
  if "API_REQUEST" in [message] {
    jdbc {
      driver_jar_path => '/usr/share/logstash/vendor/jar/jdbc/mssql-jdbc-12.2.0.jre8.jar'
      connection_string => "jdbc:sqlserver://ip:1433;databaseName=izdb;user=user;password=pass;ssl=false;trustServerCertificate=true"
      enable_event_as_json_keyword => true
      statement => [
        "MERGE INTO Transaction AS target
         USING (SELECT ? AS document_id, ? AS logLevel, ? AS timestamp) AS source
         ON target.document_id = source.document_id
         WHEN MATCHED THEN
             UPDATE SET logLevel = source.logLevel, timestamp = source.timestamp
         WHEN NOT MATCHED THEN
             INSERT (document_id, logLevel, timestamp) VALUES (source.document_id, source.logLevel, source.timestamp);",
        "document_id",
        "logLevel",
        "timestamp"
      ]
    }
  }
}

方案2：仅用增量查询避免重复（适合只读同步场景） #

如果你的ES文档不会更新，只需要同步新增数据，只需配置增量查询即可，无需修改SQL语句：

• 开启record_last_run和last_run_metadata_path，用@timestamp作为过滤条件，每次只拉取上次同步之后的文档，从根源避免重复拉取旧数据。

方案3：用fingerprint生成稳定唯一标识 #

如果不想依赖ES的_id，可以用fingerprint过滤器基于文档的核心字段（比如message+timestamp）生成稳定的哈希值，作为唯一标识：

filter {
  fingerprint {
    # 基于核心字段生成稳定哈希
    source => ["message", "timestamp"]
    target => "document_id"
    method => "SHA256"
    # 可选：添加密钥增强哈希唯一性
    key => "your-custom-secret-key"
  }
}

同样需要给document_id添加唯一约束，并配合MERGE语句实现幂等插入。

内容的提问来源于stack exchange，提问作者Ali farahzadi