elasticsearch 在单个字段中聚合一些值

Question

我有一些原始数据

{
  {
    "id":1,
    "message":"intercept_log,UDP,0.0.0.0,68,255.255.255.255,67"
  },
  {
    "id":2,
    "message":"intercept_log,TCP,172.22.96.4,52085,239.255.255.250,3702,1:"
  },
  {
    "id":3,
    "message":"intercept_log,UDP,1.0.0.0,68,255.255.255.255,67"
  },
  {
    "id":4,
    "message":"intercept_log,TCP,173.22.96.4,52085,239.255.255.250,3702,1:"
  }
}

需求

我想按消息的消息部分的值对这些数据进行分组。 这样的输出值

{
  {
    "GroupValue":"TCP",
    "DocCount":"2"
  },
  {
    "GroupValue":"UDP",
    "DocCount":"2"
  }
}

试试

• 我已尝试使用这些代码但失败了

GET systemevent*/_search
{
  "size": 0, 
  "aggs": {
    "tags": {
      "terms": {
        "field": "message.keyword",
        "include": " intercept_log[,，](.*?)[,，].*?"
      }
    }
  },
  "track_total_hits": true
}

• 现在我尝试使用管道来满足这一需求。
• “aggs”似乎只对字段进行分组。
• 有人有更好的主意吗？

链接

Terms aggregation

更新

我的场景有点特别。我从许多不同的服务器收集日志，然后将日志导入es。因此，消息字段之间存在很大差异。如果直接使用脚本语句进行分组统计，会导致分组失败或分组不准确。我尝试根据条件过滤掉一些数据，然后用脚本对操作代码进行分组（注释代码1），但是这段代码无法对正确的结果进行分组。

这是我要添加的场景：

我们团队使用es分析服务器日志，使用rsyslog将数据转发到服务器中心，然后使用logstash过滤提取数据到es。这时候ES中有一个叫message的字段，message的值就是详细的日志信息。这时，我们需要统计消息中包含一些值的数据。

注释代码1

POST systemevent*/_search
{
  "size": 0, 
  "query": {
    "bool": {
      "must": [
        {
          "match_phrase": {
            "message": {
              "query": "intercept_log"
            }
          }
        }
      ]
    }
  },
  "aggs": {
    "protocol": {
      "terms": {
        "script": "def values = /,/.split(doc['message.keyword'].value); return values.length > 1 ? values[1] : 'N/A'",
        "size": 10
      }
    }
  },
  "track_total_hits": true
}

注释代码2

POST test2/_search
{
  "size": 0,
  "aggs": {
    "protocol": {
      "terms": {
        "script": "def values = /.*,.*/.matcher( doc['host.keyword'].value ); if( name.matches() ) {return values.group(1) } else { return 'N/A' }",
        "size": 10
      }
    }
  }
}

Answer 1

解决此问题的最简单方法是利用 terms 聚合中的脚本。该脚本将简单地以逗号分隔并取第二个值。

    POST systemevent*/_search
    {
      "size": 0,
      "aggs": {
        "protocol": {
          "terms": {
            "script": "def values = /,/.split(doc['message.keyword'].value); return values.length > 1 ? values[1] : 'N/A';",
            "size": 10
          }
        }
      }
    }

使用正则表达式

POST test2/_search
{
  "size": 0,
  "aggs": {
    "protocol": {
      "terms": {
        "script": "def m = /.*proto='(.*?)'./.matcher(doc['message.keyword'].value ); if( m.matches() ) { return m.group(1) } else { return 'N/A' }"
      }
    }
  }
}

结果看起来像

  "buckets" : [
    {
      "key" : "TCP",
      "doc_count" : 2
    },
    {
      "key" : "UDP",
      "doc_count" : 2
    }
  ]

更好、更有效的方法是使用摄取管道或 Logstash 将 message 字段拆分为新字段。