如何通过Logstash每日向Elastic加载周数据并避免重复？

针对你的需求，有两种实用的方案可以实现无重复同步，结合你的场景推荐优先选择第一种：

方案一：基于文档唯一ID的幂等写入 #

核心思路是利用Elasticsearch的文档ID唯一性，让重复数据直接被跳过，无需额外维护同步状态。

实现步骤： #

1. 确定数据的唯一标识：每条环境数据必须有一个唯一键（比如自带的uuid，或者通过设备ID+数据生成时间戳组合生成，比如device_123_1699999200）。
2. 配置Logstash输出插件：
   • 在elasticsearch输出中指定document_id为这个唯一键
   • 设置action => "create"，同时开启ignore_conflicts => true，这样当Elasticsearch中已存在该ID的文档时，Logstash会自动跳过这条数据，不会覆盖也不会中断管道。

示例配置： #

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "env_data-%{+YYYY.MM.dd}"
    # 用数据中的设备ID和时间戳组合生成唯一文档ID
    document_id => "%{[device_id]}_%{[event_timestamp]}"
    action => "create"
    ignore_conflicts => true
  }
}

优势： #

• 配置简单，无需额外存储同步状态
• 首次执行会写入全部7天数据，后续每日同步时，前6天的重复数据因ID已存在会被自动跳过，仅写入新增的1天数据

方案二：基于同步时间戳的增量过滤 #

核心思路是记录上次同步的结束时间，每次仅保留时间戳大于该值的数据，过滤掉已同步过的旧记录。

实现步骤： #

1. 维护同步状态：用本地文件或Elasticsearch专用索引存储上次同步的结束时间戳（比如首次初始化时写入0，表示同步所有数据）。
2. 读取同步状态并过滤数据：在Logstash中读取该时间戳，对比每条数据的生成时间，丢弃时间戳小于等于该值的记录。
3. 更新同步状态：同步完成后，将当前时间减去1小时（匹配接口返回的时间范围逻辑）写入状态存储。

示例配置（本地文件存储状态）： #

input {
  http {
    url => "http://your-api-url/env-data"
    schedule => "0 0 * * *" # 每日凌晨执行同步
    codec => json
    # 读取上次同步时间戳
    add_field => { "last_sync_time" => "%{file:///opt/logstash/config/last_sync_time.txt}" }
  }
}

filter {
  # 转换时间戳为整数类型用于对比
  mutate {
    convert => {
      "last_sync_time" => "integer"
      "event_timestamp" => "integer"
    }
  }
  # 丢弃已同步过的旧数据
  if [event_timestamp] <= [last_sync_time] {
    drop {}
  }
}

output {
  elasticsearch {
    hosts => ["http://localhost:9200"]
    index => "env_data-%{+YYYY.MM.dd}"
  }
  # 更新同步时间戳为当前时间减1小时（匹配接口返回的最新数据截止时间）
  file {
    path => "/opt/logstash/config/last_sync_time.txt"
    content => "%{+%s}"
    write_behavior => "overwrite"
  }
}

优势： #

• 不依赖数据的唯一标识，仅需数据包含精确的生成时间戳
• 可以精确控制同步的时间范围，适合数据无天然唯一键的场景

注意事项 #

• 确保数据中的时间字段与接口返回的时间范围完全匹配，避免漏同步或重复
• 如果数据量较大，方案一的性能优于方案二，因为无需额外的状态查询或文件读写
• 若使用方案二，多实例部署时建议用Elasticsearch存储同步状态，避免本地文件的冲突问题

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