Elasticsearch如何单查询实现先算中位数再筛选大于中位数的文档

实现思路 #

利用ES管道聚合能力，在同一次查询中先计算中位数，再通过桶过滤器筛选符合条件的文档，不需要分两次请求。

可用DSL示例 #

{
  "size": 0,
  "aggs": {
    "median_calc": {
      "percentiles": {
        "field": "age",
        "percents": [50],
        "tdigest": {
          "compression": 1000 // 可调高该值提升中位数计算精度，默认100
        }
      }
    },
    "all_doc_buckets": {
      "terms": {
        "field": "_id",
        "size": 10000 // 需大于等于索引实际文档总数，最大支持默认的10000，超量可调整index.max_result_window参数
      },
      "aggs": {
        "current_age": {
          "max": {
            "field": "age"
          }
        },
        "gte_median_filter": {
          "bucket_selector": {
            "buckets_path": {
              "medianVal": "median_calc[50.0]",
              "ageVal": "current_age"
            },
            "script": "params.ageVal >= params.medianVal"
          }
        },
        "result_docs": {
          "top_hits": {
            "size": 1
          }
        }
      }
    }
  }
}

返回结果中aggregations.all_doc_buckets.buckets下每个桶的result_docs.hits.hits就是符合要求的文档。

Java Rest High Level Client 实现示例 #

// 构建中位数聚合
PercentilesAggregationBuilder medianAgg = AggregationBuilders.percentiles("median_calc")
        .field("age")
        .percentiles(50)
        .compression(1000);

// 构建文档分桶聚合
TermsAggregationBuilder docBuckets = AggregationBuilders.terms("all_doc_buckets")
        .field("_id")
        .size(10000);

// 子聚合：取当前文档age
MaxAggregationBuilder ageAgg = AggregationBuilders.max("current_age").field("age");

// 子聚合：桶过滤，保留age>=中位数的桶
BucketSelectorPipelineAggregationBuilder filterAgg = PipelineAggregatorBuilders.bucketSelector("gte_median_filter",
        new HashMap<String, String>() {{
            put("medianVal", "median_calc[50.0]");
            put("ageVal", "current_age");
        }},
        new Script(ScriptType.INLINE, "painless", "params.ageVal >= params.medianVal", Collections.emptyMap())
);

// 子聚合：返回符合条件的文档内容
TopHitsAggregationBuilder topHitsAgg = AggregationBuilders.topHits("result_docs").size(1);

// 组装聚合关系
docBuckets.subAggregation(ageAgg)
        .subAggregation(filterAgg)
        .subAggregation(topHitsAgg);

// 构建查询请求
SearchRequest searchRequest = new SearchRequest("你的索引名");
SearchSourceBuilder sourceBuilder = new SearchSourceBuilder()
        .size(0)
        .aggregation(medianAgg)
        .aggregation(docBuckets);
searchRequest.source(sourceBuilder);

// 执行查询
SearchResponse response = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);

// 解析结果
Terms allDocBuckets = response.getAggregations().get("all_doc_buckets");
for (Terms.Bucket bucket : allDocBuckets.getBuckets()) {
    TopHits topHits = bucket.getAggregations().get("result_docs");
    SearchHit hit = topHits.getHits().getAt(0);
    // 处理你的文档逻辑
    Map<String, Object> sourceAsMap = hit.getSourceAsMap();
}

注意事项 #

• 该方案适合文档总数量不大的场景（建议10万以内），如果文档量级达到百万级以上，两次查询的性能开销反而更低，因为_id分桶+top_hits在大数据量下会占用更高的堆内存。
• percentiles聚合默认是近似计算，对精度要求高可以调高compression参数值，数值越高精度越高，内存开销也越大。

内容的提问来源于stack exchange，提问作者Manonandan S K