简介

ES的聚合是一个有用的工具，能够帮助我们有效地了解我们的数据。聚合有多种类型：metric聚合，bucketing聚合，pipeline聚合，matrix聚合等。

定义一个聚合，需要遵循的模板是：

"aggregations" : {
    "<aggregation_name>" : {
        "<aggregation_type>" : {
            <aggregation_body>
        }
        [,"meta" : {  [<meta_data_body>] } ]?
        [,"aggregations" : { [<sub_aggregation>]+ } ]?
    }
    [,"<aggregation_name_2>" : { ... } ]*
}

在使用的过程中，"aggregations"可以简写成"aggs"。

查询，聚合，子聚合

这个例子是我在分析网络流量时为了统计外网IP做的。

image-20200325132608694.png

这个简单的例子分了3个区域：

query block，这部分和一般的query是一样的，此部分的搜索结果，会作为聚合部分的输入。
aggs block，聚合部分的第一层，我使用了一个基于时间的histogram桶聚合，以10分钟为步长进行bucket聚合。
sub aggs block，这部分是前一层聚合的子聚合，会在前个聚合的基础上在其分出的每个bucket中再分别进行term聚合，我在这个地方设置了第二层桶的数目为20。

得到的结果如下：

[图片上传中...(aggs.png-548586-1585124502356-0)]

可以看到，结果中有两层桶，显示了我想要的统计结果。在第二层桶里，有一个参数“sum_other_doc_count”，这个参数显示了没有在之前设置的20个桶里的文档数目。

从这个例子可以看到ES聚合时的数据流向。

aggs.png

可以看到：

查询块的结果是聚合的数据源。
同一层的聚合使用的是同一个数据源。
只有bucket 聚合才有子聚合。

在聚合中使用脚本

还是网络数据分析的场景，现有一个需求，我希望能够把内网中有访问关系的实体提出出来，然后按照会话数按Top排序，并且列出top20的peer。

应对这个需求，需要对数据进行先期处理，先将数据进行一次过滤，凡是从外网进入的，或者访问外网的首先要先过滤掉。这里我使用了一个两层嵌套bool search：

GET session_log_aliases\_search
{
  "size": 0,
  "query": {
    "bool": {
      "minimum_should_match" : 2,
      "should": [
        {
          "bool": {
            "should": [
              {
                "range": {
                  "dst_addr": {
                    "gte": "10.0.0.0",
                    "lte": "10.255.255.255"
                  }
                }
              },
              {
                "range": {
                  "dst_addr": {
                    "gte": "172.0.0.0",
                    "lte": "172.255.255.255"
                  }
                }
              },
              {
                "range": {
                  "dst_addr": {
                    "gte": "192.0.0.0",
                    "lte": "192.255.255.255"
                  }
                }
              }
            ]
          }
        },
        {
          "bool": {
            "should": [
              {
                "range": {
                  "src_addr": {
                    "gte": "10.0.0.0",
                    "lte": "10.255.255.255"
                  }
                }
              },
              {
                "range": {
                  "src_addr": {
                    "gte": "172.0.0.0",
                    "lte": "172.255.255.255"
                  }
                }
              },
              {
                "range": {
                  "src_addr": {
                    "gte": "192.0.0.0",
                    "lte": "192.255.255.255"
                  }
                }
              }
            ]
          }
        }
      ]
    }
  }

为了保证源地址和目的地址都是内网地址，需要为should块设置"minimum_should_match" : 2。

下面考虑聚合的情况。由于需要保证同源同宿的都算在一起，需要把源地址和目的地址放到一起考虑。

  "aggs": {
    "single_pair": {
      "terms": {
        "size": 40,
        "script": {
          "lang": "painless",
          "source": """
            if (doc["dst_addr"].hashCode() < doc["src_addr"].hashCode())
            {
              return doc["dst_addr"] + " " + doc["src_addr"];
            }
            else
            {
              return doc["src_addr"] + " " + doc["dst_addr"];
            }
        """
        }
      }
    }

使用term聚合，需要指定一个field来进行，然而源数据里面并没有直接可用的字段，这时候我们可以使用"painless"脚本来进行。对源地址和目的地址进行一次排序，按照固定的顺序组成一个值，由于这里只是需要一个固定的排序，直接用hash值进行对比就可以了。然后将源地址和目的地址组成一个值返回，聚合就能根据这个返回值进行了。

得到数据：

  "aggregations" : {
    "single_pair" : {
      "doc_count_error_upper_bound" : 0,
      "sum_other_doc_count" : 29788,
      "buckets" : [
        {
          "key" : "[10.21.140.3] [10.21.138.200]",
          "doc_count" : 11274
        },
        {
          "key" : "[10.21.140.3] [10.21.138.206]",
          "doc_count" : 8412
        },
        {
          "key" : "[10.21.142.8] [10.21.128.51]",
          "doc_count" : 7080
        },
        {
          "key" : "[10.21.140.3] [10.21.138.201]",
          "doc_count" : 6868
        }
        。。。