如何用ElasticSearch聚合筛选同时有login和login_error事件的用户

要实现这个需求，单纯的cardinality或terms聚合无法满足——它们只能判断用户是否存在两类事件，但无法验证事件的先后顺序。我们需要结合分桶、排序和脚本聚合来实现：

完整查询语句 #

{
  "size": 0,
  "query": {
    "bool": {
      "filter": [
        {"terms": {"event_type": ["login", "login_error"]}}
      ]
    }
  },
  "aggs": {
    "user_groups": {
      "terms": {
        "field": "user_id",
        "size": 10000  // 设置足够大的值覆盖所有6000+用户
      },
      "aggs": {
        "sorted_events": {
          "top_hits": {
            "size": 50,  // 每个用户取最多50条事件，足够覆盖登录错误后登录的场景
            "sort": [{"event_time": "asc"}]  // 按事件发生时间升序排列
          }
        },
        "qualified_flag": {
          "scripted_metric": {
            "init_script": "state.error_seen = false; state.is_qualified = false;",
            "map_script": """
              def hits = params._source.sorted_events.hits.hits;
              for (def hit : hits) {
                def event_type = hit._source.event_type;
                if (event_type == 'login_error') {
                  state.error_seen = true;
                } else if (event_type == 'login' && state.error_seen) {
                  state.is_qualified = true;
                  break;  // 找到符合条件的事件后提前终止遍历
                }
              }
            """,
            "combine_script": "return state.is_qualified ? 1 : 0;",
            "reduce_script": "return states.stream().mapToInt(Integer::intValue).sum();"
          }
        }
      }
    },
    "total_qualified_users": {
      "sum_bucket": {
        "buckets_path": "user_groups>qualified_flag"
      }
    }
  }
}

关键逻辑说明 #

1. 过滤事件范围：先用bool filter只保留login和login_error类型的事件，减少不必要的数据处理
2. 按用户分桶：通过terms聚合将数据按user_id分组，确保每个桶对应单个用户的事件
3. 事件排序：top_hits聚合获取每个用户的事件并按event_time升序排列，保证我们按事件实际发生顺序处理
4. 脚本验证顺序：scripted_metric在每个用户桶内遍历事件：
   • 初始化标记：是否见过login_error，是否符合条件
   • 遍历事件时，先标记已出现登录错误，后续如果出现登录事件则判定该用户符合条件
   • 每个桶返回1（符合）或0（不符合）
5. 统计符合条件的用户数：sum_bucket聚合将所有用户桶的结果求和，得到最终的目标数量

注意事项 #

• 确保event_time字段是日期类型，否则无法正确排序
• terms聚合的size值要大于实际用户数，避免遗漏用户
• top_hits的size可根据实际业务调整，只要能覆盖用户从登录错误到成功登录的事件序列即可
• 脚本中的字段名（如event_type、event_time）要和你索引中的实际字段名完全一致

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