Elasticsearch统计Top5接口时如何正则替换URL用户ID合并统计

方案1：查询时实时处理（无需修改现有索引，适合临时统计场景） #

你可以在聚合时新增runtime字段，用Painless脚本对原始URL做清洗，替换掉用户ID段为*之后再做聚合，修改后的查询DSL如下：

{
  "runtime_mappings": {
    "cleaned_url": {
      "type": "keyword",
      "script": {
        "source": """
          // 匹配用户ID的正则：纯字母/纯数字/字母数字混合，长度≥8（可根据实际ID长度调整）
          def idPattern = /^[a-zA-Z0-9]{8,}$/;
          def urlParts = doc['data.url.keyword'].value.splitOnToken('/');
          def cleanedParts = new ArrayList();
          for (def part : urlParts) {
            if (part == '') continue;
            // 匹配到用户ID特征就替换为*
            if (idPattern.matcher(part).matches()) {
              cleanedParts.add('*');
            } else {
              cleanedParts.add(part);
            }
          }
          emit('/' + String.join('/', cleanedParts));
        """
      }
    }
  },
  "query": {
    "bool": {
      "must": [
        {
          "range": {
            "@timestamp": {
              "from": "now-15m",
              "to": "now",
              "include_lower": true,
              "include_upper": true,
              "boost": 1
            }
          }
        }
      ],
      "adjust_pure_negative": true,
      "boost": 1
    }
  },
  "aggregations": {
    "Url": {
      "terms": {
        "field": "cleaned_url", // 改为聚合清洗后的URL字段
        "size": 5,
        "min_doc_count": 1,
        "shard_min_doc_count": 0,
        "show_term_doc_count_error": false,
        "order": [
          {
            "_count": "desc"
          },
          {
            "_key": "asc"
          }
        ]
      },
      "aggregations": {
        "Status": {
          "terms": {
            "field": "data.response.status",
            "size": 5,
            "min_doc_count": 1,
            "shard_min_doc_count": 0,
            "show_term_doc_count_error": false,
            "order": [
              {
                "_count": "desc"
              },
              {
                "_key": "asc"
              }
            ]
          }
        }
      }
    }
  }
}

• 可根据你实际的用户ID长度、特征调整正则规则，避免误替换正常的接口路径段。

方案2：写入时预处理（性能更高，适合长期高频统计场景） #

如果这类统计是高频需求，建议在数据写入Elasticsearch时就通过Ingest Pipeline做预处理，把清洗后的路径存为独立的keyword字段，后续直接聚合该字段即可，避免每次查询都做实时计算消耗性能。
预处理Pipeline示例：

PUT _ingest/pipeline/clean_url_pipeline
{
  "processors": [
    {
      "script": {
        "source": """
          def idPattern = /^[a-zA-Z0-9]{8,}$/;
          def urlParts = ctx.data.url.splitOnToken('/');
          def cleanedParts = new ArrayList();
          for (def part : urlParts) {
            if (part == '') continue;
            if (idPattern.matcher(part).matches()) {
              cleanedParts.add('*');
            } else {
              cleanedParts.add(part);
            }
          }
          ctx.data.clean_url = '/' + String.join('/', cleanedParts);
        """
      }
    }
  ]
}

配置Pipeline生效后，后续写入的文档都会新增data.clean_url字段，你可以直接用该字段做聚合，查询性能会比方案1高很多。

结果格式适配 #

聚合返回后，你只需要遍历Url聚合的buckets，把每个bucket的key和下属Status聚合的结果拼接即可得到你期望的*/payment 200:2格式。

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