如何在Elasticsearch聚合中基于脚本字段进行过滤？

我明白你想要在Elasticsearch的聚合流程里，通过自定义脚本字段来实现特定的过滤逻辑。从你给出的代码片段来看，你已经在top_hits里定义了reqd_flag这个脚本字段，但代码没写完，接下来我会帮你补全逻辑，并展示两种常见的基于脚本字段的聚合过滤方式。

核心思路 #

脚本字段可以帮我们动态计算出一个字段值，之后我们可以基于这个值在文档层面（分桶前过滤）或者桶层面（分桶后过滤）做筛选。下面分别给出两种场景的实现方案：

方案1：分桶前基于脚本过滤文档 #

如果想只让符合脚本字段条件的文档进入后续的terms聚合，可以在filters聚合里添加一个基于脚本的过滤条件，替代原来的match_all：

{
  "size": 0,
  "query": {
    "bool": {
      "must": [
        {
          "range": {
            "datetime": {
              "gte": "2017-04-01T00:00:00.000Z",
              "lte": "2018-03-31T23:59:59.999Z"
            }
          }
        }
      ]
    }
  },
  "aggs": {
    "filtered_docs": {
      "filters": {
        "filters": {
          "reqd_docs": {
            "script": {
              "lang": "painless",
              "inline": "// 这里写你的判断逻辑，比如判断_source里的某个字段满足条件
                        params['_source'].some_field == 'required' || params['_source'].priority > 3"
            }
          }
        }
      },
      "aggs": {
        "jobs": {
          "terms": {
            "field": "job_num",
            "size": 200000
          },
          "aggs": {
            "latest_job": {
              "top_hits": {
                "size": 1,
                "sort": [{"rec_date": "desc"}],
                "script_fields": {
                  "reqd_flag": {
                    "script": {
                      "lang": "painless",
                      "inline": "// 和上面的过滤逻辑保持一致，返回true/false
                                params['_source'].some_field == 'required' || params['_source'].priority > 3"
                    }
                  }
                }
              }
            }
          }
        }
      }
    }
  }
}

方案2：分桶后过滤不符合条件的桶 #

如果希望先完成job_num的分桶，再过滤掉那些最新文档不满足reqd_flag条件的桶，可以用bucket_selector管道聚合：

{
  "size": 0,
  "query": {
    "bool": {
      "must": [
        {
          "range": {
            "datetime": {
              "gte": "2017-04-01T00:00:00.000Z",
              "lte": "2018-03-31T23:59:59.999Z"
            }
          }
        }
      ]
    }
  },
  "aggs": {
    "all_match": {
      "filters": {
        "filters": {
          "all": {
            "match_all": {}
          }
        }
      },
      "aggs": {
        "jobs": {
          "terms": {
            "field": "job_num",
            "size": 200000
          },
          "aggs": {
            "latest_job": {
              "top_hits": {
                "size": 1,
                "sort": [{"rec_date": "desc"}],
                "script_fields": {
                  "reqd_flag": {
                    "script": {
                      "lang": "painless",
                      "inline": "params['_source'].some_field == 'required'"
                    }
                  }
                }
              }
            },
            // 提取脚本字段的值，用于后续桶过滤
            "reqd_flag_value": {
              "max": {
                "script": {
                  "lang": "painless",
                  "inline": "params['_source'].some_field == 'required' ? 1 : 0"
                }
              }
            },
            // 过滤掉reqd_flag为false的桶
            "filter_valid_jobs": {
              "bucket_selector": {
                "buckets_path": {
                  "flag": "reqd_flag_value"
                },
                "script": "params.flag == 1"
              }
            }
          }
        }
      }
    }
  }
}

关键说明 #

• 脚本逻辑要根据你的实际业务需求调整，比如判断字段是否存在、字段值是否符合特定规则等；
• 方案1适合提前过滤掉不需要的文档，减少后续聚合的计算量；方案2适合需要保留所有分桶但只展示符合条件的桶的场景；
• 确保你的Elasticsearch集群允许使用inline脚本（如果是较新版本，可能需要在elasticsearch.yml中配置script.allowed_types: inline）。

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