更新具有相同 ID 的弹性搜索文档答案

【问题标题】：update elastic-search document with the same ID更新具有相同 ID 的弹性搜索文档
【发布时间】：2019-07-28 08:32:36
【问题描述】：

每个人。我是麋鹿的新手，我有一个关于logstash的问题。我有一些服务，每个服务都有 4 或 6 个日志；这意味着弹性文档可能有 4 或 6 个日志。我想阅读这些日志，如果它们具有相同的 id，请将它们放在一个弹性文档中。我必须指定所有日志都具有唯一的“id”，并且每个请求和引用该请求的每个日志都具有相同的 id。每个日志都有特定的类型。我想把每个具有相同 id 和类型的日志放在一起；像这样：

    {
      "_id":"123",
      "Type1":{},
      "Type2":[{},{}],
      "Type3":[{},{}],
      "Type4":{}
    }

同一请求集的每个日志：其中一些必须在同一组中。因为它们的类型相同。看上面的例子。 Type2 是 Json 数组，有 2 个 json。我想使用 logstash 来读取每条日志并将它们分类。想象一下，我们的文档现在就像下面的 JSON：

    {
      "_id": "123",
      "Type1":{},
      "Type2":[{},{}],
      "Type3":{}
    }

现在有一个新的日志到达，它的 id 为 123，它的类型是 Type4。文档必须像这样更新：

    {
      "_id": "123",
      "Type1":{},
      "Type2":[{},{}],
      "Type3":{},
      "Type4":{}
    }

再次，我有新的日志，ID 为 123，类型为 Type3。文档更新如下：

    {
      "_id": "123",
      "Type1":{},
      "Type2":[{},{}],
      "Type3":[{},{}],
      "Type4":{}
    }

我尝试使用脚本，但没有成功。：

    {
      "id": 1,
        "Type2": {}
    }

脚本是：

input {
    stdin {
        codec => json_lines
    }
}
output {
    elasticsearch {
      hosts => ["XXX.XXX.XXX.XXX:9200"]
      index => "ss"
      document_id => "%{requestId}"
      action => "update" # update if possible instead of overwriting 
      document_type => "_doc"
      script_lang => "painless"
      scripted_upsert => true
      script_type => "inline"
      script => 'if (ctx._source.Type3 == null) { ctx._source.Type3 = new ArrayList() } if(!ctx._source.Type3.contains("%{Type3}")) { ctx._source.Type3.add("%{Type3}")}'
    }
}

现在我的问题是这种脚本格式只是一种类型；如果它适用于多种类型，它会是什么样子？还有一个问题。我有一些日志，它们没有 id，或者它们有 id，但没有类型。我想把这些日志放到elastic里面，怎么办？

【问题讨论】：

标签： elasticsearch logstash kibana elk

【解决方案1】：

您可以查看用于 logstash 的 aggregate 过滤器插件。或者如您所说，如果某些日志没有 id，那么您可以使用 fingerprint 过滤器插件创建一个 id，您可以使用它来更新 elasticsearch 中的文档。例如：

input {
    stdin {
        codec => json_lines
    }
}
filter {
  fingerprint {
     source => "message"
     target => "[@metadata][id]"
     method => "MURMUR3"
  }
}
output {
   elasticsearch {
      hosts => ["XXX.XXX.XXX.XXX:9200"]
      index => "ss"
      document_id => "%{[@metadata][id]}"
      action => "update" # update if possible instead of overwriting
    }
}

【讨论】：