管道聚集不是直接从索引中读取文档，而是在其他聚集的基础上再进行聚集运算。所以管道聚集可以理解为是在聚集结果上再次做聚集运算，比如求聚集结果中多个桶中某一指标的平均值、最大值等。要实现这样的目的，管道聚集都会包含一个名为buckets_path的参数，用于指定访问其他桶中指标值的路径。
buckets_ path参数的值由三部分组成，即聚集名称、指标名称和分隔符。聚集名称与聚集名称之间的分隔符是“>”，而聚集名称与指标名称之间的分隔符使用“.”。
按管道聚集运算来源分类，管道聚集可以分为基于父聚集结果和基于兄弟聚集结果两类。前者使用父聚集的结果并将运算结果添加到父聚集结果中，后者则使用兄弟聚集的结果并且结果会展示在自己的聚集结果中。
基于兄弟聚集基于兄弟聚集的管道聚集包括avg_bucket、max_bucket、min_bucket、sum_bucket、stats_bucket、extended_ stats_ bucket、percentiles_bucket七种。如果将它们名称中的bucket去除，它们就与之前介绍的部分指标聚集同名了。事实上，它们不仅在名称上接近，而且在功能上也类似，只是聚集运算的范围由整个文档变成了另一个聚集结果。
以avg_bucket为例， 它的作用是计算兄弟聚集结果中某一指标的平均值：
POST /kibana_sample_data_flights/_search?filter_path=aggregations
{
"aggs": {
"carriers": {
"terms": {
"field": "Carrier",
"size": 10
},
"aggs": {
"carrier_stat": {
"stats": {
"field": "AvgTicketPrice"
}
}
}
},
"all_stat": {
"avg_bucket": {
"buckets_path": "carriers>carrier_stat.avg"
}
}
}
}
输出结果：
{
"aggregations" : {
"carriers" : {
"doc_count_error_upper_bound" : 0,
"sum_other_doc_count" : 0,
"buckets" : [
{
"key" : "Logstash Airways",
"doc_count" : 3331,
"carrier_stat" : {
"count" : 3331,
"min" : 100.37113189697266,
"max" : 1199.72900390625,
"avg" : 624.5819742276625,
"sum" : 2080482.5561523438
}
},
{
"key" : "JetBeats",
"doc_count" : 3274,
"carrier_stat" : {
"count" : 3274,
"min" : 101.0330810546875,
"max" : 1199.642822265625,
"avg" : 627.4573726292857,
"sum" : 2054295.4379882812
}
},
{
"key" : "Kibana Airlines",
"doc_count" : 3234,
"carrier_stat" : {
"count" : 3234,
"min" : 100.14596557617188,
"max" : 1199.109130859375,
"avg" : 630.8681507004435,
"sum" : 2040227.5993652344
}
},
{
"key" : "ES-Air",
"doc_count" : 3220,
"carrier_stat" : {
"count" : 3220,
"min" : 100.0205307006836,
"max" : 1199.5123291015625,
"avg" : 630.235816375069,
"sum" : 2029359.3287277222
}
}
]
},
"all_stat" : {
"value" : 628.2858284831152
}
}
}
例中，最外层包含有两个名称分别为carriers和all_stat的聚集，这两个聚集就是兄弟关系。carries聚集是一个针对Carrier 字段的terms聚集，Carrier字段保存的是航班承运航空公司，所以这个聚集的作用是按航空公司将航班分桶。在这个聚集中嵌套了一个名为carrier_stat的聚集，它是一个针对AvgTicketPrice字段的stats聚集，会按桶计算票价的最小值、平均值等统计数据。
all_ stat聚集则是一个avg_bucket管道聚集，在它的Path参数中指定了运算平均值的路径" carriers>carrier_stat.avg "，即从兄弟聚集中查找carrier_stat指标聚集， 然后再用其中的avg字段参与平均值计算。所以all_stat这计算出来的是四个航空公司平均票价的平均值，实际上就是所有航班的平均票价。
尽管示例针对avg_bucket管道聚集的检索，但使用其余六种基于兄弟的管道聚集的关键字值替换后，它们就变成了另一种合法的管道聚集了。
基于父聚集基于父聚集的管道聚集包括moving_avg、moving_fn、bucket_script、bucket_selector、bucket_sort、derivative、cumulative_sum、serial_diff八种。
滑动窗口moving_avg和 moving_fn这两种管道聚集的运算机制相同，都是基于滑动窗口（Siding Window）算法对父聚集的结果做新的聚集运算。滑动窗口算法使用一个具有固定宽度的窗口滑过一组数据， 在滑动的过程中对落在窗口内的数据做运算。moving_avg 管道聚集是对落在窗口内的父聚集结果做平均值运算，而moving_fn管道聚集则可以对落在窗口内的父聚集结果做各种自定义的运算。由于moving_avg管道可以使用moving_fn管道聚集实现，所以moving_avg在Elaticearch版本6.4.0中已经被废止。
由于使用滑动窗口运算时每次移动1个位置， 这就要求moving_avg和moving_fn所在聚集桶与桶间隔必须固定，所以这两种管道聚集只能在histogam和date_histogam聚集中使用：
POST /kibana_sample_data_flights/_search?filter_path=aggregations
{
"aggs": {
"week_price": {
"date_histogram": {
"field": "timestamp",
"calendar_interval": "week"
},
"aggs": {
"avg_price": {
"avg": {
"field": "AvgTicketPrice"
}
},
"smooth_price": {
"moving_fn": {
"buckets_path": "avg_price",
"window": 10,
"script": "MovingFunctions.unweightedAvg(values)"
}
}
}
}
}
}
输出结果：
{
"aggregations" : {
"day_price" : {
"buckets" : [
{
"key_as_string" : "2021-11-01T00:00:00.000Z",
"key" : 1635724800000,
"doc_count" : 2202,
"avg_price" : {
"value" : 637.2777728185558
},
"smooth_price" : {
"value" : null
}
},
{
"key_as_string" : "2021-11-08T00:00:00.000Z",
"key" : 1636329600000,
"doc_count" : 2177,
"avg_price" : {
"value" : 635.5069089441821
},
"smooth_price" : {
"value" : 637.2777728185558
}
},
{
"key_as_string" : "2021-11-15T00:00:00.000Z",
"key" : 1636934400000,
"doc_count" : 2142,
"avg_price" : {
"value" : 618.9750309506941
},
"smooth_price" : {
"value" : 636.3923408813689
}
},
{
"key_as_string" : "2021-11-22T00:00:00.000Z",
"key" : 1637539200000,
"doc_count" : 2187,
"avg_price" : {
"value" : 624.024733087708
},
"smooth_price" : {
"value" : 630.5865709044773
}
},
{
"key_as_string" : "2021-11-29T00:00:00.000Z",
"key" : 1638144000000,
"doc_count" : 2188,
"avg_price" : {
"value" : 620.4439880050296
},
"smooth_price" : {
"value" : 628.9461114502849
}
},
{
"key_as_string" : "2021-12-06T00:00:00.000Z",
"key" : 1638748800000,
"doc_count" : 2163,
"avg_price" :
"value" : 633.13114766594
},
"smooth_price" : {
"value" : 627.2456867612339
}
}
]
}
}
}
在示例中，最外层的父聚集week_price是1个date_histogam桶型聚集，它根据文档的timestamp字段按周将文档分桶。day_price聚集包含avg_price和smooth_price两个子聚集，其中avg_price聚集是一个求AvgTicketPrice字段在1个桶内平均值的avg聚集，而smooth_price则是一个使用滑动窗口做平均值平滑的管道聚集，窗口宽度由参数window设置为10，默认值为5。通过返回结果比较avg_price与smooth_price就会发现，后者由于经过了滑动窗口运算，数据变化要平滑得多。
moving_fn聚集包含一个用于指定运算脚本的script参数，在脚本中可以通过values访问buckets_path参数指定的指标值。moving_fn还内置了一个MovingFunctions类，包括多个运算函数：
- max()：最大值
- min()：最小值
- sum()：求和
- stdDev()：标准偏差
- unweightedAvg()：无加权平均值
- linearWeightedAvg()：线性加权移动平均值
- ewma()：指数加权移动平均值
- holt()：二次指数加权移动平均值
- holtWinters()：三次指数加权移动平均值
之前的管道聚集会对父聚集结果中落在窗口内的多个桶做聚集运算，而bucket_script、bucket_selector、bucket_sort这三个管道聚集则会针对父聚集结果中的每一个桶做单独的运算。其中，bucket_script会对每个桶执行一段脚本，运算结果会添加到父聚集的结果中，bucket_selector同样也是执行一段脚本，但它执行的结果一定是布尔类型，并且决定当前桶是否出现在父聚集的结果中；bucket_sort则根据每桶中的具体指标值决定桶的次序。
下面通过示例来说明这三种管道聚集的具体用法:
POST /kibana_sample_data_flights/_search?filter_path=aggregations
{
"aggs": {
"date_price_diff": {
"date_histogram": {
"field": "timestamp",
"fixed_interval": "1d"
},
"aggs": {
"stat_price_day": {
"stats": {
"field": "AvgTicketPrice"
}
},
"diff": {
"bucket_script": {
"buckets_path": {
"max_price": "stat_price_day.max",
"min_price": "stat_price_day.min"
},
"script": "params.max_price - params.min_price"
}
},
"gt990": {
"bucket_selector": {
"buckets_path": {
"max_price": "stat_price_day.max",
"min_price": "stat_price_day.min"
},
"script": "params.max_price - params.min_price > 990"
}
},
"sort_by": {
"bucket_sort": {
"sort": [
{
"diff": {
"order": "desc"
}
}
]
}
}
}
}
}
}
输出结果：
{
"aggregations" : {
"date_price_diff" : {
"buckets" : [
{
"key_as_string" : "2021-11-28T00:00:00.000Z",
"key" : 1638057600000,
"doc_count" : 241,
"stat_price_day" : {
"count" : 241,
"min" : 201.05032348632812,
"max" : 1199.4913330078125,
"avg" : 781.9357850640641,
"sum" : 188446.52420043945
},
"diff" : {
"value" : 998.4410095214844
}
},
{
"key_as_string" : "2021-11-06T00:00:00.000Z",
"key" : 1636156800000,
"doc_count" : 346,
"stat_price_day" : {
"count" : 346,
"min" : 202.06753540039062,
"max" : 1199.72900390625,
"avg" : 779.62087906854,
"sum" : 269748.82415771484
},
"diff" : {
"value" : 997.6614685058594
}
},
{
"key_as_string" : "2021-11-07T00:00:00.000Z",
"key" : 1636243200000,
"doc_count" : 214,
"stat_price_day" : {
"count" : 214,
"min" : 200.0950469970703,
"max" : 1195.72509765625,
"avg" : 773.6875114084404,
"sum" : 165569.12744140625
},
"diff" : {
"value" : 995.6300506591797
}
},
{
"key_as_string" : "2021-12-11T00:00:00.000Z",
"key" : 1639180800000,
"doc_count" : 339,
"stat_price_day" : {
"count" : 339,
"min" : 205.7853240966797,
"max" : 1196.7706298828125,
"avg" : 796.9324892657345,
"sum" : 270160.113861084
},
"diff" : {
"value" : 990.9853057861328
}
},
{
"key_as_string" : "2021-12-05T00:00:00.000Z",
"key" : 1638662400000,
"doc_count" : 223,
"stat_price_day" : {
"count" : 223,
"min" : 208.52029418945312,
"max" : 1199.109130859375,
"avg" : 799.2644480889154,
"sum" : 178235.97192382812
},
"diff" : {
"value" : 990.5888366699219
}
}
]
}
}
}
在示例中同时应用这三种管道聚集，它们的聚集名称分别为diff、gt990和sort_by。最外层的date_price_diff聚集是一个以天为固定间隔的date_histogram聚集，其中又嵌套了包括上述三个聚集在内的子聚集。
其中，stat_price_day是一个根据AvgTicketPrice字段生成统计数据的stats聚集。diff是一个bucket_script管道聚集，它的作用是向最终聚集结果中添加代表最大值与最小值之差的diff字段。它通过buckets_path定义了两个参数max_price和 min_price，并在script参数中通过脚本计算了这两个值的差作为最终结果，而这个结果将出现在整个聚集结果中。
gt990是一个bucket_selector管道聚集，它的作用是筛选哪些桶可以出现在最终的聚集结果中。它也在buckets_path中定义了相同的参数，不同的是它的script参数运算的不是差值，而是差值是否大于990。如果差值大 990即运算结果为true，那么当前桶将被选取到结果中，否则当前桶将不能在结果中出现。
sort_by是一个bucket_sort管道聚集，它的作用是给最终的聚集结果排序。它通过sort参数接收一组排序对象，在示例中是使用diff聚集的结果按倒序排序。
所以示例整体的运算效果就是将那些票价最大值与最小值大于990的桶选取出来，并在桶中添加diff字段保存最大值与最小值的差值，并按diff字段值降序排列。
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）