数据去重

Beats 框架保证至少一次交付，以确保在事件发送到支持确认的输出（例如 Elasticsearch、Logstash、Kafka 和 Redis）时不会丢失数据。如果一切按计划进行，这很好。但是，如果 Filebeat 在处理过程中关闭，或者在事件被确认之前连接丢失，则可能会得到重复数据。

当输出被阻塞时，Filebeat 中的重试机制会尝试重新发送事件，直到输出确认它们。如果输出接收到事件，但无法确认它们，则数据可能会多次发送到输出。因为文档 ID 通常由 Elasticsearch 在从 Beats 接收到数据之后设置，所以重复的事件会被索引为新文档。

与其允许 Elasticsearch 设置文档 ID，不如在 Beats 中设置 ID。ID 存储在 Beats 的 @metadata._id 字段中，用于在索引期间设置文档 ID。这样，如果 Beats 多次向 Elasticsearch 发送相同的事件，Elasticsearch 将覆盖现有文档，而不是创建新文档。

@metadata._id 字段与事件一起传递，以便您可以在 Filebeat 发布事件之后但在 Elasticsearch 接收到事件之前使用它来设置文档 ID。例如，请参阅 Logstash 管道示例。

有几种方法可以在 Beats 中设置文档 ID

对于此示例，假设您已使用前面描述的方法之一将文档 ID 存储在 Beats @metadata._id 字段中。当您通过 Logstash 发送 Beats 数据到 Elasticsearch 时，为了保留 ID，请在 Logstash 管道中设置 document_id 字段

input {
  beats {
    port => 5044
  }
}

output {
  if [@metadata][_id] {
    elasticsearch {
      hosts => ["https://127.0.0.1:9200"]
      document_id => "%{[@metadata][_id]}" 
      index => "%{[@metadata][beat]}-%{[@metadata][version]}"
    }
  } else {
    elasticsearch {
      hosts => ["https://127.0.0.1:9200"]
      index => "%{[@metadata][beat]}-%{[@metadata][version]}"
    }
  }
}

当 Elasticsearch 索引文档时，它会将文档 ID 设置为指定的值，从而保留从 Beats 传递的 ID。