简介
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
}
。。。
