Logstash MySQL 组合多行

Question

我一直在使用 logstash，并且能够将我的一些 MySQL 表插入到 Elastic 搜索中。

现在，我需要将 MySQL Query statememt 中的多行合并到一个文档中。例如，假设我有一个 mySQL 查询，我有以下结果，

1   Group1  Company1    User1
1   Group1  Company1    User2
1   Group2  Company2    User3
1   Group2  Company2    User4

我想在 Elastic Search 中为 Group1 和 Group2 创建 2 个 Group 文档。在 Group1 中，我需要添加 User1 和 User2。在 Group2 中，我需要添加 User3 和 User4。

这是否可以使用 LogStash 并与一些过滤器结合使用？如果是这样，任何人都可以指出我正确的方向。

Answer 1

您可以使用logstash-filter-aggregate。过滤器采用task_id，在您的情况下为组字段，并聚合具有相同 task_id 值的行。因此，group1 的事件将被聚合，然后当过滤器检测到新的 group 值时，group1 的聚合用户数据将作为单个 Logstash 事件推送到您的 elasticsearch 文档。然后，它使用 group2 重新开始该过程。

确保将 logstash 过滤器设置为 1（-w 1 标志或在配置文件中），并且行的排序方式是所有 group1 事件都进来，然后是所有 group2 等。否则事件可能被乱序处理，并可能产生意外的结果/数据丢失。

filter {
  aggregate {
    task_id => "%{group}"
    code => "
      map['users'] ||= []
      map['users'].push(event.get('user'))
    "
    push_previous_map_as_event => true
    timeout_tags => ['aggregated']
  }

  if "aggregated" not in [tags] {
    drop {}
  }
}

结果将如下所示：

"_source": {
  "group": "group1",
  "users": ["user1", "user2"]
}