7.去重复数据.md

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

是什么导致Elasticsearch中的重复？

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

如何避免重复？

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

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

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

• add_id 处理器

  当您的数据没有natural key field，并且您无法从现有字段派生唯一密钥时，请使用add_id处理器。

  此示例为每个事件生成一个唯一ID，并将其添加到@metadata._id字段：

  processors:
    - add_id: ~

• fingerprint 处理器

  使用 fingerprint 处理器从一个或多个现有字段导出唯一密钥。

  此示例使用field1和field2的值来派生它添加到@metadata._id字段：

  processors:
    - fingerprint:
        fields: ["field1", "field2"]
        target_field: "@metadata._id"

• decode_json_fields 处理器

  在解码包含 natural key field 的JSON字符串时，使用 decode_json_fields 处理器中的document_id设置。

  对于此示例，假设消息字段包含JSON字符串{"myid": "100", "text": "Some text"}。此示例从JSON字符串中获取myid的值并将其存储在@metadata._id字段：

  processors:
    - decode_json_fields:
        document_id: "myid"
        fields: ["message"]
        max_depth: 1
        target: ""

  生成的文档ID为 100。

• JSON input settings

  如果正在获取JSON格式的数据，并且数据具有natural key field，请使用json.document_id输入设置。

  此示例从JSON文档中获取key1的值并将其存储在@metadata._id字段中：

  filebeat.inputs:
  - type: log
    paths:
      - /path/to/json.log
    json.document_id: "key1"

Logstash pipline example

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

input {
  beats {
    port => 5044
  }
}

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

将 Elasticsearch output 中的document_id字段设置为存储在@metadata._id。