Logstash Elasticsearch过滤器插件聚合使用与'aggregation_fields'报错排查

错误原因 #

你遇到的Unknown setting 'aggregation_fields' for elasticsearch错误，核心原因是Logstash的Elasticsearch过滤器插件根本没有aggregation_fields这个合法配置项。这个参数并非插件官方支持的设置，所以Logstash启动时会将其识别为无效配置并抛出错误。

正确实现聚合的方法 #

要在Elasticsearch过滤器中完成聚合操作，核心思路是在查询模板里定义Elasticsearch的聚合DSL，再通过插件的aggregation_results配置将聚合结果注入到当前处理的事件中。具体步骤如下：

1. 修正Logstash过滤器配置 #

先移除无效的aggregation_fields参数，添加aggregation_results来指定存储聚合结果的字段名，示例配置如下：

if [type] == "mytype" {
  elasticsearch {
    hosts => ["192.168.1.1:9200"]
    index => "my_index"
    query_template => "/logstash/my_query.json"
    # 指定聚合结果存入当前事件的`my_agg_results`字段
    aggregation_results => "my_agg_results"
  }
}

2. 在查询模板中编写聚合DSL #

修改你的my_query.json文件，在Elasticsearch查询语句中加入聚合逻辑。比如你要对seqnumber字段做最大值聚合（可根据需求替换为min/avg/terms等其他聚合类型），示例内容如下：

{
  "query": {
    "match_all": {}
  },
  "aggs": {
    "find_seqnumber": {
      "max": {
        "field": "seqnumber"
      }
    }
  }
}

这里的aggs块是Elasticsearch的聚合定义，find_seqnumber是聚合的自定义名称，max是你需要的聚合类型。

3. 验证聚合结果 #

当Logstash处理事件时，聚合结果会被存入你指定的my_agg_results字段中。你可以通过stdout输出插件查看结果，示例输出配置：

output {
  stdout {
    codec => rubydebug
  }
}

最终事件中会包含类似这样的聚合结果：

"my_agg_results" => {
  "find_seqnumber" => {
    "value" => 12345
  }
}

额外提示 #

• 确保你的Elasticsearch版本与Logstash版本兼容，避免因版本差异导致聚合DSL不支持的问题。
• 如果需要基于当前事件的字段动态生成聚合条件，可以在查询模板中使用%{field_name}的Logstash变量语法，比如：

  {
    "query": {
      "match": {
        "related_field": "%{current_event_field}"
      }
    },
    "aggs": {
      "find_seqnumber": {
        "max": {
          "field": "seqnumber"
        }
      }
    }
  }
  

内容的提问来源于stack exchange，提问作者Yaser