如何通过Elasticsearch聚合获取指定数量最旧文档的日期极值

我需要从Elasticsearch索引中筛选出N个最旧的文档，并直接通过聚合计算这部分文档的date字段最小和最大值，而非全量符合条件文档的极值。后续要处理数千个最旧文档，希望由Elasticsearch完成计算，避免本地处理大量数据。

已尝试的方法及问题 #

1. 初始查询：获取N个最旧文档 #

通过排序+限制size可以拿到目标文档，但需要本地提取日期后计算极值：

### sorted search limited to N result
POST _search?filter_path=hits.total,hits.hits._source.date
{
  "_source": [ "date" ],
  "size": 3,
  "sort": { "date": "asc" },
  "query": { "bool": { "filter": { "term": { "queue.name": "messages.in.dlq" }}}}
}

// 返回结果
{
  "hits": {
    "total": 254,
    "hits": [
      { "_source": { "date": "2023-10-31T10:51:32.185+0100" } },
      { "_source": { "date": "2023-10-31T11:13:15.919+0100" } },
      { "_source": { "date": "2023-10-31T14:39:37.559+0100" } }
    ]
  }
}

2. 直接聚合：得到全量文档极值 #

添加min/max聚合后，计算的是所有符合条件文档的日期极值，而非目标N个最旧文档的：

### aggregated search
POST errormessages-search/_search?filter_path=aggregations.*.value_as_string
{
  "_source": [ "date" ],
  "size": 3,
  "sort": { "date": "asc" },
  "query": { "term": { "queue.name": "messages.in.dlq" }},
  "aggs": {
    "date_min": { "min": { "field": "date" }},
    "date_max": { "max": { "field": "date" }}
  }
}

// 返回结果（date_max为全量文档最大值，不符合需求）
{
  "aggregations": {
    "date_min": { "value_as_string": "2023-10-31T09:51:32.185Z" },
    "date_max": { "value_as_string": "2024-01-08T09:03:10.565Z" } //!\ overall date_max not the oldest 3
  }
}

3. top_hits+子聚合：不支持嵌套 #

尝试用top_hits先获取N个文档，再嵌套聚合计算极值，但Elasticsearch不允许top_hits添加子聚合，直接报错：

### top_hits then min max aggregation
POST _search?filter_path=aggregations.oldest_agg.aggregations
{
  "size": 0,
  "query": { "bool": { "filter": {"term": { "queue.name": "messages.in.dlq" }}}},
  "aggs": {
    "oldest_agg": {
      "top_hits": {
        "size": 3,
        "sort": { "date": "asc" }
      },
      "aggs": {
        "date_min": { "min": { "field": "date" }},
        "date_max": { "max": { "field": "date" }}
      }
    }
  }
}

// 错误返回
{
  "error": {
    "type": "aggregation_initialization_exception",
    "reason": "Aggregator [oldest_agg] of type [top_hits] cannot accept sub-aggregations"
  },
  "status": 500
}

4. sampler聚合：结果不符合排序要求 #

使用sampler聚合尝试限制样本，但sampler是按shard取前N个（非按日期排序），导致采样的文档不是最旧的目标集合，聚合结果错误：

### sampler aggregation
GET errormessages-search/_search?filter_path=**.date,**.doc_count,**.date_*.value_as_string
{
  "_source": [ "date" ],
  "size": 3,
  "sort": { "date": "asc" },
  "query": { "bool": { "filter": { "term": { "queue.name": "messages.in.dlq" }}}},
  "aggs": {
    "sample": {
      "sampler": {
        "shard_size": 3
      },
      "aggs": {
        "date_min": { "min": { "field": "date" }},
        "date_max": { "max": { "field": "date" }}
      }
    }
  }
}

// 返回结果（采样的文档不是最旧的3个）
{
  "hits": {
    "hits": [
      { "_source": { "date": "2024-01-02T20:57:22.812+0100" } },
      { "_source": { "date": "2024-01-02T20:57:22.812+0100" } },
      { "_source": { "date": "2024-01-02T20:57:22.901+0100" } }
    ]
  },
  "aggregations": {
    "sample": {
      "doc_count": 18,
      "date_min": { "value_as_string": "2024-01-02T19:57:22.812Z" },
      "date_max": { "value_as_string": "2024-01-03T01:22:02.233Z" }
    }
  }
}

可行方案：使用scripted_metric聚合 #

scripted_metric允许自定义聚合逻辑，可实现"先收集前N个最旧文档的日期，再计算极值"的需求。核心逻辑是在每个shard上收集并排序日期、保留前N个，最后合并所有shard的结果计算最终min/max：

POST errormessages-search/_search?filter_path=aggregations.*
{
  "size": 0,
  "query": {
    "bool": {
      "filter": {
        "term": { "queue.name": "messages.in.dlq" }
      }
    }
  },
  "aggs": {
    "oldest_dates_stats": {
      "scripted_metric": {
        "init_script": "state.dates = []; state.limit = params.limit;",
        "map_script": """
          // 收集当前shard的文档日期，保持升序并只保留前limit个
          def date = doc['date'].value;
          state.dates.add(date);
          state.dates.sort();
          if (state.dates.size() > state.limit) {
            state.dates.remove(state.dates.size() - 1);
          }
        """,
        "combine_script": """
          // 合并当前shard内的结果，确保只保留前limit个最旧日期
          state.dates.sort();
          if (state.dates.size() > state.limit) {
            return [dates: state.dates.subList(0, state.limit)];
          }
          return [dates: state.dates];
        """,
        "reduce_script": """
          // 合并所有shard的结果，最终取前limit个最旧日期并计算极值
          def allDates = [];
          for (shardResult in states) {
            allDates.addAll(shardResult.dates);
          }
          allDates.sort();
          if (allDates.size() > params.limit) {
            allDates = allDates.subList(0, params.limit);
          }
          return [
            min_date: allDates.size() > 0 ? allDates[0].toString() : null,
            max_date: allDates.size() > 0 ? allDates[allDates.size() - 1].toString() : null,
            count: allDates.size()
          ];
        """,
        "params": {
          "limit": 3 // 替换为你需要的N值
        }
      }
    }
  }
}

返回结果示例 #

{
  "aggregations": {
    "oldest_dates_stats": {
      "min_date": "2023-10-31T09:51:32.185Z",
      "max_date": "2023-10-31T13:39:37.559Z",
      "count": 3
    }
  }
}

该方案可高效处理数千个文档的场景，所有计算由Elasticsearch完成，无需本地传输大量数据。

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