根据文档ID在多个logstash上组合输出答案

【问题标题】：Combined output on multiple logstash according to document ID根据文档ID在多个logstash上组合输出
【发布时间】：2020-01-20 01:48:08
【问题描述】：

我有来自 RabbitMq 的 json 日志。在我的例子中，我使用三个不同的 logstash 服务来使用三个不同的查询，它会使用相同的索引输出到 elasticsearch。假设我们有日志 A、日志 B 和日志 C。这些日志的结构如下：

Example of Log A

Example of Log B

Example of Log C

使用这些日志并将这些日志发送到弹性的配置文件与：

Logstash.conf

问题 => 在发送到具有相同索引的弹性之前，如何根据 document_id 组合这三个不同的日志？

也许 JSON 输出如下所示：

Result

【问题讨论】：

标签： json logstash elastic-stack

【解决方案1】：

弹性搜索端可以使用document_id合并日志，也可以使用elasticsearch更新查询执行更新功能，与已有记录合并。

下面的代码会检查elasticsearch上是否有带有文档id的文档，如果没有文档，则会创建一个新文档。如果存在具有文档 ID 的文档，它将执行更新查询。

为了澄清，请查看 docs-update-by-query

试试下面的代码：



elasticsearch {

                hosts => [ "xxxxxx"]
                index => "combined"

                action => "update"

                doc_as_upsert => true

                document_id => "%{[header][serialNum]}"

}

【讨论】：

嗨蒂维娅！谢谢你回复我……你的回答对我很有帮助。所有日志都合并在一个索引上！不幸的是，它用相同的字段和属性替换了所有日志。我想知道如何将它与相同的索引和 ID 组合成一个对象或数组。再次感谢您

【解决方案2】：

我看到一个建议使用更新选项的答案，

使用 elasticsearch "action => update" 会疯狂地降低你的性能。随着流量的增加，您会看到 ES 集群工作非常努力。

“update”会首先让ES搜索文档，合并文档，删除现有文档，添加这个新的合并文档。

这是很多处理！

我的回答：

使用聚合过滤器。有一个 task_id 字段可以对其进行排序。您可以将其设置为

aggregate {
 task_id => %{[header][serialNum]}
 ...
}

https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html

【讨论】：