Elasticsearch实现按user_id分组取event_type=1的最新记录

问题分析 #

现有查询存在两个核心问题：

• 排序逻辑错误：top_hits使用user_id降序排序，无法获取每组的最新记录；应改为按id（对应MySQL的max(id)逻辑）或event_date降序排序。
• 缺少过滤逻辑：未对组内最新记录的event_type进行筛选，导致返回所有user_id组的最新记录，不符合仅保留event_type=1的需求。

修正后的查询语句 #

以下查询基于MySQL的max(id)逻辑（用id判断最新记录），如果需要用event_date判断，只需将排序字段替换为event_date即可：

GET test_analytic_report/_search
{
  "from": 0,
  "size": 0,
  "query": {
    "bool": {
      "must": [
        {
          "range": {
            "event_date": {
              "gte": "2022-10-01",
              "lte": "2023-02-06"
            }
          }
        }
      ]
    }
  },
  "aggs": {
    "group_by_user": {
      "terms": {
        "field": "user_id"
      },
      "aggs": {
        "latest_record": {
          "top_hits": {
            "size": 1,
            "_source": ["user_id", "event_date", "event_type", "id"],
            "sort": {
              "id": "desc" // 对应MySQL的max(id)，若用event_date则改为"event_date": "desc"
            }
          }
        },
        // 提取最新记录的event_type用于过滤
        "event_type_value": {
          "script_metric": {
            "init_script": "state.event_type = null",
            "map_script": "state.event_type = params._source.event_type",
            "combine_script": "return state.event_type",
            "reduce_script": "return states[0]"
          }
        },
        // 只保留event_type=1的分组
        "filter_valid_groups": {
          "bucket_selector": {
            "buckets_path": {
              "type": "event_type_value"
            },
            "script": "params.type == 1"
          }
        }
      }
    }
  }
}

关键修正点说明 #

1. 修正排序逻辑：将top_hits的排序字段改为id（或event_date）降序，确保获取每组的最新记录。
2. 新增脚本聚合提取event_type：通过script_metric聚合提取每组最新记录的event_type值。
3. 添加bucket_selector过滤：使用bucket_selector聚合仅保留event_type=1的分组，过滤掉不符合条件的user_id组。

替代方案（客户端过滤） #

如果你的Elasticsearch版本不支持script_metric，也可以先获取所有组的最新记录，再在客户端对结果进行过滤，仅保留event_type=1的记录。但这种方式会返回多余数据，建议优先使用上述聚合级别的过滤方案。

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