Elasticsearch使用篇 - 管道聚合

管道聚合

基于前一次聚合的结果，进行二次聚合统计。

从结构上可以分为兄弟级（Sibling）管道聚合和父级（Parent）管道聚合两种方式。

• 兄弟级管道聚合：在同一聚合级别上可以产生新的聚合。

GET kibana_sample_data_logs/_search
{
  "size": 0,
  "aggs": {
    "count_per_day": {
      "date_histogram": {
        "field": "@timestamp",
        "calendar_interval": "day"
      }
    },
    "total_bytes_of_download": {
      "sum": {
        "field": "bytes"
      }
    }
  }
}

• 父级管道聚合：由父聚合提供输出，子聚合能够产生新的桶，然后可以添加到父桶中。

GET  kibana_sample_data_logs/_search
{
  "size": 0,
  "aggs": {
    "count_per_day": {
      "date_histogram": {
        "field": "@timestamp",
        "calendar_interval": "day"
      },
      "aggs": {
        "total_bytes_per_day": {
          "sum": {
            "field": "bytes"
          }
        }
      }
    }
  }
}

max_bucket、min_bucket、avg_bucket、sum_bucket

基于兄弟级管道聚合的方式，从多个分桶中获取指定数值指标的聚合结果（最大值 / 最小值 / 平均值 / 和）。

• buckets_path：（必须）目标桶的路径指向。
• gap_policy：分桶数据出现空值或者缺失值的处理策略，默认 skip。gap_policy 详细说明
  • skip：跳过空值或者缺失值，不参与聚合计算。
  • insert_zeros：将空值或者缺失值当作 0 来参与聚合计算。
  • keep_values：如果提供的指标值是一个空值或者缺失值（null、NAN）则跳过，不参与聚合计算；否则使用该指标值参与聚合计算。
• format：指定数值的输出格式，比如 #、#0.00。默认 null。

1、统计每个出发地国家的最长的航班飞行时间，并且获取航班飞行时间最长的国家以及对应的航班飞行时间。

GET kibana_sample_data_flights/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "terms_OriginCountry": {
      "terms": {
        "field": "OriginCountry",
        "size": 20
      },
      "aggs": {
        "max_FlightTimeMin": {
          "max": {
            "field": "FlightTimeMin"
          }
        }
      }
    },
    "pipeline-max-bucket": {
      "max_bucket": {
        "buckets_path": "terms_OriginCountry>max_FlightTimeMin"
      }
    }
  }
}

截取部分的聚合结果如下：

"pipeline-max-bucket" : {
  "value" : 1902.9019775390625,
  "keys" : [
    "AR"
  ]
}

2、获取出发地国家中各个出发地城市的航班飞行时间的最大值，然后统计出发地城市中航班飞行时间最长的飞行时长以及对应的出发地城市名称。

GET kibana_sample_data_flights/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "terms_OriginCountry": {
      "terms": {
        "field": "OriginCountry",
        "size": 20
      },
      "aggs": {
        "terms_OriginCityName": {
          "terms": {
            "field": "OriginCityName",
            "size": 20
          },
          "aggs": {
            "max_FlightTimeMin": {
              "max": {
                "field": "FlightTimeMin"
              }
            }
          }
        },
        "Max_Bucket_OriginCityName_FlightTimeMin": {
          "max_bucket": {
            "buckets_path": "terms_OriginCityName>max_FlightTimeMin"
          }
        }
      }
    }
  }
}

截取以出发地国家维度的第一个分桶中的管道聚合的结果。

"Max_Bucket_OriginCityName_FlightTimeMin" : {
  "value" : 1559.6236572265625,
  "keys" : [
    "Rome"
  ]
}

3、在 2 的基础上获取出发地国家中航班最长的飞行时间以及对应的出发地国家名称。

GET kibana_sample_data_flights/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "terms_OriginCountry": {
      "terms": {
        "field": "OriginCountry",
        "size": 20
      },
      "aggs": {
        "terms_OriginCityName": {
          "terms": {
            "field": "OriginCityName",
            "size": 20
          },
          "aggs": {
            "max_FlightTimeMin": {
              "max": {
                "field": "FlightTimeMin"
              }
            }
          }
        },
        "Max_Bucket_OriginCityName_FlightTimeMin": {
          "max_bucket": {
            "buckets_path": "terms_OriginCityName>max_FlightTimeMin"
          }
        }
      }
    },
    "Max_Bucket_OriginCountry_FlightTimeMin": {
      "max_bucket": {
        "buckets_path": "terms_OriginCountry>Max_Bucket_OriginCityName_FlightTimeMin"
      }
    }
  }
}

在 2 的基础上额外输出出发地国家纬度的管道聚合结果。

"Max_Bucket_OriginCountry_FlightTimeMin" : {
  "value" : 1902.9019775390625,
  "keys" : [
    "AR"
  ]
}

stats_bucket

基于兄弟级管道聚合的方式，从多个分桶中获取指定数值指标的统计聚合结果。

• buckets_path：（必须）目标桶的路径指向。
• gap_policy：分桶数据出现空值或者缺失值的处理策略，默认 skip。gap_policy 详细说明
  • skip：跳过空值或者缺失值，不参与聚合计算。
  • insert_zeros：将空值或者缺失值当作 0 来参与聚合计算。
  • keep_values：如果提供的指标值是一个空值或者缺失值（null、NAN）则跳过，不参与聚合计算；否则使用该指标值参与聚合计算。
• format：指定数值的输出格式，比如 #、#0.00。默认 null。
• sigma：标准差的倍数，默认 2。用于确定一个数据点是否属于异常值。

1、统计出发地国家中各个出发地城市的航班飞行时间的平均时长，然后统计这些平均时长在出发地国家纬度下的各项指标。

GET kibana_sample_data_flights/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "terms_OriginCountry": {
      "terms": {
        "field": "OriginCountry",
        "size": 20
      },
      "aggs": {
        "terms_OriginCityName": {
          "terms": {
            "field": "OriginCityName",
            "size": 20
          },
          "aggs": {
            "avg_FlightTimeMin": {
              "avg": {
                "field": "FlightTimeMin"
              }
            }
          }
        },
        "Stats_Bucket_OriginCityName_FlightTimeMin": {
          "stats_bucket": {
            "buckets_path": "terms_OriginCityName>avg_FlightTimeMin"
          }
        }
      }
    }
  }
}

截取部分的聚合结果如下：

"Stats_Bucket_OriginCityName_FlightTimeMin" : {
  "count" : 15,
  "min" : 226.4979310909907,
  "max" : 472.0975369329038,
  "avg" : 378.1233526619374,
  "sum" : 5671.850289929062
}

extended_stats_bucket

基于兄弟级管道聚合的方式，从多个分桶中获取指定数值指标的拓展统计聚合结果。

• buckets_path：（必须）目标桶的路径指向。
• gap_policy：分桶数据出现空值或者缺失值的处理策略，默认 skip。gap_policy 详细说明
  • skip：跳过空值或者缺失值，不参与聚合计算。
  • insert_zeros：将空值或者缺失值当作 0 来参与聚合计算。
  • keep_values：如果提供的指标值是一个空值或者缺失值（null、NAN）则跳过，不参与聚合计算；否则使用该指标值参与聚合计算。
• format：指定数值的输出格式，比如 #、#0.00。默认 null。
• sigma：标准差的倍数，默认 2。用于确定一个数据点是否属于异常值。

1、统计出发地国家中各个出发地城市的航班飞行时间的平均时长，然后统计这些平均时长在出发地城市纬度下的各项指标。

GET kibana_sample_data_flights/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "terms_OriginCountry": {
      "terms": {
        "field": "OriginCountry",
        "size": 20
      },
      "aggs": {
        "terms_OriginCityName": {
          "terms": {
            "field": "OriginCityName",
            "size": 20
          },
          "aggs": {
            "avg_FlightTimeMin": {
              "avg": {
                "field": "FlightTimeMin"
              }
            }
          }
        },
        "Extended_Stats_Bucket_OriginCityName_FlightTimeMin": {
          "extended_stats_bucket": {
            "buckets_path": "terms_OriginCityName>avg_FlightTimeMin"
          }
        }
      }
    }
  }
}

截取部分的聚合结果如下：

"Extended_Stats_Bucket_OriginCityName_FlightTimeMin" : {
  "count" : 15,
  "min" : 226.4979310909907,
  "max" : 472.0975369329038,
  "avg" : 378.1233526619374,
  "sum" : 5671.850289929062,
  "sum_of_squares" : 2264926.8781446246,
  "variance" : 8017.855381337739,
  "variance_population" : 8017.855381337739,
  "variance_sampling" : 8590.559337147579,
  "std_deviation" : 89.54247808352045,
  "std_deviation_population" : 89.54247808352045,
  "std_deviation_sampling" : 92.68527033540755,
  "std_deviation_bounds" : {
    "upper" : 557.2083088289783,
    "lower" : 199.03839649489652,
    "upper_population" : 557.2083088289783,
    "lower_population" : 199.03839649489652,
    "upper_sampling" : 563.4938933327526,
    "lower_sampling" : 192.75281199112231
  }
}

cumulative_sum

[ˈkjuːmjəleɪtɪv]，累计的；累积的

基于父级管道聚合的方式，对直方图聚合、日期直方图聚合中的相关数值指标进行累积聚合统计。

外层的直方图聚合、日期直方图聚合的 min_doc_count 必须设置为 0。

• buckets_path：（必须）目标桶的路径指向。
• format：指定数值的输出格式，比如 #、#0.00。默认 null。

1、统计用户每天的订单消费总额以及每天增长的累计消费金额。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "date_histogram_order_date": {
      "date_histogram": {
        "field": "order_date",
        "calendar_interval": "day",
        "format": "yyyy-MM-dd"
      },
      "aggs": {
        "stats_taxful_total_price": {
          "stats": {
            "field": "taxful_total_price"
          }
        },
        "pipeline_cumulative_sum": {
          "cumulative_sum": {
            "buckets_path": "stats_taxful_total_price.sum"
          }
        }
      }
    }
  }
}

截取部分的聚合结果如下：

{
  "key_as_string" : "2022-07-14",
  "key" : 1657756800000,
  "doc_count" : 146,
  "stats_taxful_total_price" : {
    "count" : 146,
    "min" : 18.984375,
    "max" : 230.0,
    "avg" : 72.45569349315069,
    "sum" : 10578.53125
  },
  "pipeline_cumulative_sum" : {
    "value" : 10578.53125
  }
},
{
  "key_as_string" : "2022-07-15",
  "key" : 1657843200000,
  "doc_count" : 153,
  "stats_taxful_total_price" : {
    "count" : 153,
    "min" : 22.984375,
    "max" : 220.0,
    "avg" : 68.2875816993464,
    "sum" : 10448.0
  },
  "pipeline_cumulative_sum" : {
    "value" : 21026.53125
  }
},
{
  "key_as_string" : "2022-07-16",
  "key" : 1657929600000,
  "doc_count" : 143,
  "stats_taxful_total_price" : {
    "count" : 143,
    "min" : 18.984375,
    "max" : 250.0,
    "avg" : 71.91247814685315,
    "sum" : 10283.484375
  },
  "pipeline_cumulative_sum" : {
    "value" : 31310.015625
  }
}

cumulative_cardinality

[kɑːdɪ’nælɪtɪ]，基数

基于父级管道聚合的方式，对直方图聚合、日期直方图聚合中的相关数值指标进行累积基数聚合统计。

外层的直方图聚合、日期直方图聚合的 min_doc_count 必须设置为 0。

• buckets_path：（必须）目标桶的路径指向。
• format：指定数值的输出格式，比如 #、#0.00。默认 null。

1、统计每天下单的用户数以及累计的用户数。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "date_histogram_order_date": {
      "date_histogram": {
        "field": "order_date",
        "calendar_interval": "day",
        "format": "yyyy-MM-dd"
      },
      "aggs": {
        "cardinality_customer_id": {
          "cardinality": {
            "field": "customer_id"
          }
        },
        "pipeline_cumulative_cardinality": {
          "cumulative_cardinality": {
            "buckets_path": "cardinality_customer_id"
          }
        }
      }
    }
  }
}

截取部分的聚合结果如下：

{
  "key_as_string" : "2022-07-14",
  "key" : 1657756800000,
  "doc_count" : 146,
  "cardinality_customer_id" : {
    "value" : 42
  },
  "pipeline_cumulative_cardinality" : {
    "value" : 42
  }
},
{
  "key_as_string" : "2022-07-15",
  "key" : 1657843200000,
  "doc_count" : 153,
  "cardinality_customer_id" : {
    "value" : 44
  },
  "pipeline_cumulative_cardinality" : {
    "value" : 46
  }
},
{
  "key_as_string" : "2022-07-16",
  "key" : 1657929600000,
  "doc_count" : 143,
  "cardinality_customer_id" : {
    "value" : 45
  },
  "pipeline_cumulative_cardinality" : {
    "value" : 46
  }
}

moving_avg

移动平均值聚合。基于父级管道聚合的方式，在指定的数据序列中滑动一个窗口，统计窗口内部的平均值。

• buckets_path：（必须）目标桶的路径指向。

• window：滑动窗口的大小。默认 5。

• gap_policy：分桶数据出现空值或者缺失值的处理策略，默认 skip。gap_policy 详细说明

  • skip：跳过空值或者缺失值，不参与聚合计算。
  • insert_zeros：将空值或者缺失值当作 0 来参与聚合计算。
  • keep_values：如果提供的指标值是一个空值或者缺失值（null、NAN）则跳过，不参与聚合计算；否则使用该指标值参与聚合计算。

• model：指定移动平均值聚合的模型，默认 simple。每个模型对于窗口内部的值的加权方式不同。

  • simple：简单模型。它会计算窗口内部的所有值的和，然后除以窗口大小。简单模型不执行随时间变化的加权，这意味着该模型下的移动平均值往往滞后于真实数据。
  • linear：线性模型。它会对旧的数据点分配线性权重，以此来减少数据平均值的滞后性。
  • ewma：单指数模型。它会对旧的数据点分配指数权重。权重衰减的速度可以通过 alpha 参数控制。默认 0.3。alpha 参数支持 0 到 1 之间的浮点数，值越小可以使权重缓慢衰减，提供了更好的平滑效果；值越大可以使权重快速衰减，减少了旧值对于移动平均值的影响，虽然平滑度较低，但是使移动平均值更紧密地跟踪数据。该模型可以最小化。
  • holt：双指数模型。模型内部计算两个值：水平、趋势。基于数据的趋势，可以预测未来走向。alpha 参数对应水平衰减值，默认 0.3。beta 参数对应趋势衰减值，默认 0.1。alpha 参数、beta 参数都支持 0 到 1 之间的浮点数。该模型可以最小化
  • holt_winters：三指数模型。模型内部计算三个值：水平、趋势、季节性。基于数据的季节性变化，可以预测未来走向。alpha 参数对应水平衰减值，默认 0.3。beta 参数对应趋势衰减值，默认 0.1。gamma 参数对应季节衰减值，默认 0.3。alpha 参数、beta 参数、gamma 参数都支持 0 到 1 之间的浮点数。period 参数对应周期，默认 1。type 参数控制季节变化如何作用于数据，支持 add、mult。该模型可以最小化。

• settings：指定模型的相关参数。

• predict：指定预测的数量（会添加到序列的末尾）。每个移动平均值模型都支持预测模式，基于当前平滑的移动平均值推测未来数据。根据模型和参数的不同，预测结果的准确性也会有所不同。比如：predict: 10。

• minimize：指定模型是否开启最小化。最小化是一个调整参数的过程，直到模型生成的预测与输出数据紧密匹配。对于 ewma、holt 模型，该参数默认 false，并且用处不大；对于 holt_winters 模型，该参数默认 true，有助于提高预测的准确性。比如：minimize: true。

对 simple 模型采用窗口大小为 3，进行举例说明。

分桶序号		分桶值			移动平均值
	1		 10
	2		 20				 10
	3		 30				(10 + 20) / 2
	4		 40				(10 + 20 + 30) / 3
	5		 50				(20 + 30 + 40) / 3

1、统计用户每天的订单消费金额，以及连续三天的平均消费金额。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "date_histogram_order_date": {
      "date_histogram": {
        "field": "order_date",
        "calendar_interval": "day",
        "format": "yyyy-MM-dd"
      },
      "aggs": {
        "stats_taxful_total_price": {
          "stats": {
            "field": "taxful_total_price"
          }
        },
        "pipeline_moving_avg": {
          "moving_avg": {
            "buckets_path": "stats_taxful_total_price.sum",
            "window": 3,
            "model": "simple"
          }
        }
      }
    }
  }
}

截取部分的聚合结果如下：

{
  "key_as_string" : "2022-07-14",
  "key" : 1657756800000,
  "doc_count" : 146,
  "stats_taxful_total_price" : {
    "count" : 146,
    "min" : 18.984375,
    "max" : 230.0,
    "avg" : 72.45569349315069,
    "sum" : 10578.53125
  }
},
{
  "key_as_string" : "2022-07-15",
  "key" : 1657843200000,
  "doc_count" : 153,
  "stats_taxful_total_price" : {
    "count" : 153,
    "min" : 22.984375,
    "max" : 220.0,
    "avg" : 68.2875816993464,
    "sum" : 10448.0
  },
  "pipeline_moving_avg" : {
    "value" : 10578.53125
  }
},
{
  "key_as_string" : "2022-07-16",
  "key" : 1657929600000,
  "doc_count" : 143,
  "stats_taxful_total_price" : {
    "count" : 143,
    "min" : 18.984375,
    "max" : 250.0,
    "avg" : 71.91247814685315,
    "sum" : 10283.484375
  },
  "pipeline_moving_avg" : {
    "value" : 10513.265625
  }
}

换成 ewma 模型看下效果。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "order_date_histogram": {
      "date_histogram": {
        "field": "order_date",
        "calendar_interval": "day",
        "format": "yyyy-MM-dd"
      },
      "aggs": {
        "taxful_total_price_sum": {
          "sum": {
            "field": "taxful_total_price"
          }
        },
        "pipeline_moving_avg": {
          "moving_avg": {
            "buckets_path": "taxful_total_price_sum",
            "window": 5,
            "model": "ewma",
            "settings": {
              "alpha": 0.5
            }
          } 
        }
      }
    }
  }
}

moving_fn

移动函数聚合。基于父级管道聚合的方式，在指定的数据序列中滑动一个窗口，使用自定义脚本的方式计算出窗口内部的值。移动函数聚合内置了一些常见的函数。

• buckets_path：（必须）目标桶的路径指向。

• window：（必须）滑动窗口的大小。

• script：（必须）对每个窗口内部的数据执行的脚本。

• gap_policy：分桶数据出现空值或者缺失值的处理策略，默认 skip。gap_policy 详细说明

  • skip：跳过空值或者缺失值，不参与聚合计算。
  • insert_zeros：将空值或者缺失值当作 0 来参与聚合计算。
  • keep_values：如果提供的指标值是一个空值或者缺失值（null、NAN）则跳过，不参与聚合计算；否则使用该指标值参与聚合计算。

• shift：指定窗口的起始位置向右移动多少位。默认 0，即窗口内部的聚合计算不包括当前桶。该值每增加一，则窗口的起始位置向右移动一位。如果指定窗口内部的聚合计算包括当前桶，则可以将 shift 设置为 1。

移动函数聚合内置了一些常见函数，如下：

• MovingFunctions.max(values)：获取最大值（忽略 null、NaN 值，如果窗口为空或者窗口内部的值都是 null、NaN，则返回 NaN ）。
• MovingFunctions.min(values)：获取最小值（忽略 null、NaN 值，如果窗口为空或者窗口内部的值都是 null、NaN，则返回 NaN ）。
• MovingFunctions.sum(values)：获取总和（忽略 null、NaN 值，如果窗口为空或者窗口内部的值都是 null、NaN，则返回 0.0 ）。
• MovingFunctions.stdDev(values, 平均值)：获取标准差（忽略 null、NaN 值，如果窗口为空或者窗口内部的值都是 null、NaN，则返回 0.0 ）。
• MovingFunctions.unweightedAvg(values)：使用 simple 模型获取平均值。
• MovingFunctions.linearWeightedAvg(values)：使用 linear 模型获取平均值。
• MovingFunctions.ewma(values, alpha)：使用 ewma 模型获取平均值。
• MovingFunctions.holt(values, alpha, beta)：使用 holt 模型获取平均值。
• MovingFunctions.holtWinters(values, alpha, beta, gamma, period, multiplicative)：使用 holt_winters 模型获取平均值。multiplicative：布尔值，true 表示使用乘法计算；false 表示使用加法计算

1、统计每天的订单消费金额，以及连续五天的消费金额总和。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "date_histogram_order_date": {
      "date_histogram": {
        "field": "order_date",
        "calendar_interval": "day",
        "format": "yyyy-MM-dd"
      },
      "aggs": {
        "stats_taxful_total_price": {
          "stats": {
            "field": "taxful_total_price"
          }
        },
        "pipeline_moving_fn": {
          "moving_fn": {
            "buckets_path": "stats_taxful_total_price.sum",
            "window": 3,
            "script": """
              MovingFunctions.sum(values)
            """,
            "shift": 1
          }
        }
      }
    }
  }
}

截取部分的聚合结果如下：

{
  "key_as_string" : "2022-07-14",
  "key" : 1657756800000,
  "doc_count" : 146,
  "stats_taxful_total_price" : {
    "count" : 146,
    "min" : 18.984375,
    "max" : 230.0,
    "avg" : 72.45569349315069,
    "sum" : 10578.53125
  },
  "pipeline_moving_fn" : {
    "value" : 10578.53125
  }
},
{
  "key_as_string" : "2022-07-15",
  "key" : 1657843200000,
  "doc_count" : 153,
  "stats_taxful_total_price" : {
    "count" : 153,
    "min" : 22.984375,
    "max" : 220.0,
    "avg" : 68.2875816993464,
    "sum" : 10448.0
  },
  "pipeline_moving_fn" : {
    "value" : 21026.53125
  }
},
{
  "key_as_string" : "2022-07-16",
  "key" : 1657929600000,
  "doc_count" : 143,
  "stats_taxful_total_price" : {
    "count" : 143,
    "min" : 18.984375,
    "max" : 250.0,
    "avg" : 71.91247814685315,
    "sum" : 10283.484375
  },
  "pipeline_moving_fn" : {
    "value" : 31310.015625
  }
}

bucket_script

分桶脚本聚合。基于父级管道聚合的方式，对多个分桶的数值类型的指标使用脚本方式进行聚合统计。

• buckets_path：（必须）目标桶的路径指向。
• script：（必须）自定义脚本。
• gap_policy：分桶数据出现空值或者缺失值的处理策略，默认 skip。gap_policy 详细说明
  • skip：跳过空值或者缺失值，不参与聚合计算。
  • insert_zeros：将空值或者缺失值当作 0 来参与聚合计算。
  • keep_values：如果提供的指标值是一个空值或者缺失值（null、NAN）则跳过，不参与聚合计算；否则使用该指标值参与聚合计算。
• format：指定数值的输出格式，比如 #、#0.00。默认 null。

1、统计每天的订单消费总额以及商品的总数，然后统计平均每件商品的消费金额。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "date_histogram_order_date": {
      "date_histogram": {
        "field": "order_date",
        "calendar_interval": "day",
        "format": "yyyy-MM-dd"
      },
      "aggs": {
        "stats_taxful_total_price": {
          "stats": {
            "field": "taxful_total_price"
          }
        },
        "stats_total_quantity": {
          "stats": {
            "field": "total_quantity"
          }
        },
        "pipeline-script": {
          "bucket_script": {
            "buckets_path": {
              "total_price": "stats_taxful_total_price.sum",
              "total_quantity": "stats_total_quantity.sum"
            },
            "script": """
              params.total_price / params.total_quantity
            """
          }
        }
      }
    }
  }
}

截取部分的聚合如下：

{
  "key_as_string" : "2022-07-14",
  "key" : 1657756800000,
  "doc_count" : 146,
  "stats_taxful_total_price" : {
    "count" : 146,
    "min" : 18.984375,
    "max" : 230.0,
    "avg" : 72.45569349315069,
    "sum" : 10578.53125
  },
  "stats_total_quantity" : {
    "count" : 146,
    "min" : 2.0,
    "max" : 4.0,
    "avg" : 2.1780821917808217,
    "sum" : 318.0
  },
  "pipeline-script" : {
    "value" : 33.2658215408805
  }
}

bucket_selector

分桶过滤器聚合。基于父级管道聚合的方式，使用脚本方式过滤出符合条件的数值类型的分桶数据参与聚合统计。脚本需要返回布尔值。如果脚本语言是 expression，则脚本可以返回数值，0 被视为 false，其它值被视为 true。

• buckets_path：（必须）目标桶的路径指向。
• script：（必须）自定义脚本。
• gap_policy：分桶数据出现空值或者缺失值的处理策略，默认 skip。gap_policy 详细说明
  • skip：跳过空值或者缺失值，不参与聚合计算。
  • insert_zeros：将空值或者缺失值当作 0 来参与聚合计算。
  • keep_values：如果提供的指标值是一个空值或者缺失值（null、NAN）则跳过，不参与聚合计算；否则使用该指标值参与聚合计算。
• format：指定数值的输出格式，比如 #、#0.00。默认 null。

1、统计每天的订单消费金额的相关指标，并筛选出总金额大于13000的日期。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "date_histogram_order_date": {
      "date_histogram": {
        "field": "order_date",
        "calendar_interval": "day",
        "format": "yyyy-MM-dd"
      },
      "aggs": {
        "stats_taxful_total_price": {
          "stats": {
            "field": "taxful_total_price"
          }
        },
        "pipeline-script": {
          "bucket_selector": {
            "buckets_path": {
              "total_price": "stats_taxful_total_price.sum"
            },
            "script": """
              params.total_price > 13000
            """
          }
        }
      }
    }
  }
}

截取部分的聚合结果如下：

{
  "key_as_string" : "2022-07-22",
  "key" : 1658448000000,
  "doc_count" : 163,
  "stats_taxful_total_price" : {
    "count" : 163,
    "min" : 18.984375,
    "max" : 393.0,
    "avg" : 83.1910467791411,
    "sum" : 13560.140625
  }
}

bucket_sort

分桶排序聚合。使用父级管道聚合的方式，对多个分桶进行排序。可以不指定字段或者指定多个字段进行排序。分桶之间可以根据 _key、_count 或者子聚合的方式进行排序。

• sort：指定用于排序的字段列表。

• from：指定从第几个开始截断。默认 0。

• size：指定返回多少个桶。默认返回所有桶。

• gap_policy：分桶数据出现空值或者缺失值的处理策略，默认 skip。gap_policy 详细说明

  • skip：跳过空值或者缺失值，不参与聚合计算。

  • insert_zeros：将空值或者缺失值当作 0 来参与聚合计算。

  • keep_values：如果提供的指标值是一个空值或者缺失值（null、NAN）则跳过，不参与聚合计算；否则使用该指标值参与聚合计算。

1、统计每天订单的消费金额中最高的两天的消费金额的相关指标。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "date_histogram_order_date": {
      "date_histogram": {
        "field": "order_date",
        "calendar_interval": "day",
        "format": "yyyy-MM-dd"
      },
      "aggs": {
        "stats_taxful_total_price": {
          "stats": {
            "field": "taxful_total_price"
          }
        },
        "pipeline-sort": {
          "bucket_sort": {
            "sort": [
              {
                "stats_taxful_total_price.sum": {
                  "order": "desc"
                }
              }
            ],
            "size": 2
          }
        }
      }
    }
  }
}

截取部分的聚合结果如下：

{
  "key_as_string" : "2022-07-22",
  "key" : 1658448000000,
  "doc_count" : 163,
  "stats_taxful_total_price" : {
    "count" : 163,
    "min" : 18.984375,
    "max" : 393.0,
    "avg" : 83.1910467791411,
    "sum" : 13560.140625
  }
},
{
  "key_as_string" : "2022-08-07",
  "key" : 1659830400000,
  "doc_count" : 165,
  "stats_taxful_total_price" : {
    "count" : 165,
    "min" : 18.984375,
    "max" : 225.0,
    "avg" : 79.36732954545455,
    "sum" : 13095.609375
  }
}

2、统计前三天的消费金额的相关指标。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "date_histogram_order_date": {
      "date_histogram": {
        "field": "order_date",
        "calendar_interval": "day",
        "format": "yyyy-MM-dd"
      },
      "aggs": {
        "stats_taxful_total_price": {
          "stats": {
            "field": "taxful_total_price"
          }
        },
        "pipeline-sort": {
          "bucket_sort": {
            "sort": [
              "_key"
            ],
            "size": 3
          }
        }
      }
    }
  }
}

截取部分的聚合结果如下：

{
  "key_as_string" : "2022-07-14",
  "key" : 1657756800000,
  "doc_count" : 146,
  "stats_taxful_total_price" : {
    "count" : 146,
    "min" : 18.984375,
    "max" : 230.0,
    "avg" : 72.45569349315069,
    "sum" : 10578.53125
  }
},
{
  "key_as_string" : "2022-07-15",
  "key" : 1657843200000,
  "doc_count" : 153,
  "stats_taxful_total_price" : {
    "count" : 153,
    "min" : 22.984375,
    "max" : 220.0,
    "avg" : 68.2875816993464,
    "sum" : 10448.0
  }
},
{
  "key_as_string" : "2022-07-16",
  "key" : 1657929600000,
  "doc_count" : 143,
  "stats_taxful_total_price" : {
    "count" : 143,
    "min" : 18.984375,
    "max" : 250.0,
    "avg" : 71.91247814685315,
    "sum" : 10283.484375
  }
}

deriative

[dɪ’rɪvətɪv]，导数

使用父级管道聚合的方式，从父级的直方图或者日期直方图聚合中获取指定数值指标并计算出导数。

父级的直方图或者日期直方图聚合的 min_doc_count 参数必须设置为 0。

可以用来获取后一个桶相较于前一个桶在相关数值指标上的变化情况。

• buckets_path：（必须）目标桶的路径指向。
• gap_policy：分桶数据出现空值或者缺失值的处理策略，默认 skip。 gap_policy详细说明
  • skip：跳过空值或者缺失值，不参与聚合计算。
  • insert_zeros：将空值或者缺失值当作 0 来参与聚合计算。
  • keep_values：如果提供的指标值是一个空值或者缺失值则跳过，不参与聚合计算；否则使用该指标值参与聚合计算。
• format：指定数值的输出格式，比如 #、#0.00。默认 null。
• unit：指定导数值的单位。具体使用参考例子3。

1、统计每天的订单销售总额以及每天的增长情况。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "date_histogram_order_date": {
      "date_histogram": {
        "field": "order_date",
        "calendar_interval": "day",
        "format": "yyyy-MM-dd"
      },
      "aggs": {
        "sum_taxful_total_price": {
          "sum": {
            "field": "taxful_total_price"
          }
        },
        "deriative_sum_taxful_total_price": {
          "derivative": {
            "buckets_path": "sum_taxful_total_price"
          }
        }
      }
    }
  }
}

截取部分的聚合结果如下：

{
  "key_as_string" : "2022-07-14",
  "key" : 1657756800000,
  "doc_count" : 146,
  "sum_taxful_total_price" : {
    "value" : 10578.53125
  }
},
{
  "key_as_string" : "2022-07-15",
  "key" : 1657843200000,
  "doc_count" : 153,
  "sum_taxful_total_price" : {
    "value" : 10448.0
  },
  "deriative_sum_taxful_total_price" : {
    "value" : -130.53125
  }
},
{
  "key_as_string" : "2022-07-16",
  "key" : 1657929600000,
  "doc_count" : 143,
  "sum_taxful_total_price" : {
    "value" : 10283.484375
  },
  "deriative_sum_taxful_total_price" : {
    "value" : -164.515625
  }
}

2、基于 1 的结果，统计每天增长情况的趋势。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "date_histogram_order_date": {
      "date_histogram": {
        "field": "order_date",
        "calendar_interval": "day",
        "format": "yyyy-MM-dd"
      },
      "aggs": {
        "sum_taxful_total_price": {
          "sum": {
            "field": "taxful_total_price"
          }
        },
        "deriative_sum_taxful_total_price": {
          "derivative": {
            "buckets_path": "sum_taxful_total_price"
          }
        },
        "second_deriative_sum_taxful_total_price": {
          "derivative": {
            "buckets_path": "deriative_sum_taxful_total_price"
          }
        }
     }
  }
}

截取部分的聚合如下：

{
  "key_as_string" : "2022-07-14",
  "key" : 1657756800000,
  "doc_count" : 146,
  "sum_taxful_total_price" : {
    "value" : 10578.53125
  }
},
{
  "key_as_string" : "2022-07-15",
  "key" : 1657843200000,
  "doc_count" : 153,
  "sum_taxful_total_price" : {
    "value" : 10448.0
  },
  "deriative_sum_taxful_total_price" : {
    "value" : -130.53125
  }
},
{
  "key_as_string" : "2022-07-16",
  "key" : 1657929600000,
  "doc_count" : 143,
  "sum_taxful_total_price" : {
    "value" : 10283.484375
  },
  "deriative_sum_taxful_total_price" : {
    "value" : -164.515625
  },
  "second_deriative_sum_taxful_total_price" : {
    "value" : -33.984375
  }
}

3、统计每月的订单销售总额以及每月的增长情况（包括平均每天的增长情况）。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "date_histogram_order_date": {
      "date_histogram": {
        "field": "order_date",
        "calendar_interval": "month",
        "format": "yyyy-MM-dd"
      },
      "aggs": {
        "sum_taxful_total_price": {
          "sum": {
            "field": "taxful_total_price"
          }
        },
        "deriative_sum_taxful_total_price": {
          "derivative": {
            "buckets_path": "sum_taxful_total_price",
            "unit": "day"
          }
        }
      }
    }
  }
}

截取部分的聚合结果如下：

{
  "key_as_string" : "2022-07-01",
  "key" : 1656633600000,
  "doc_count" : 2678,
  "sum_taxful_total_price" : {
    "value" : 199112.57421875
  }
},
{
  "key_as_string" : "2022-08-01",
  "key" : 1659312000000,
  "doc_count" : 1997,
  "sum_taxful_total_price" : {
    "value" : 151771.5546875
  },
  "deriative_sum_taxful_total_price" : {
    "value" : -47341.01953125,
    "normalized_value" : -1527.129662298387
  }
}

serial_diff

使用父级管道聚合的方式，从父级的直方图或者日期直方图聚合中获取指定数值指标并计算出指定时间段的差异。

可以用来获取后一个桶相较于前一个桶在相关数值指标上在指定时间段的变化情况。

• buckets_path：（必须）目标桶的路径指向。
• gap_policy：分桶数据出现空值或者缺失值的处理策略，默认 skip。 gap_policy详细说明
  • skip：跳过空值或者缺失值，不参与聚合计算。
  • insert_zeros：将空值或者缺失值当作 0 来参与聚合计算。
  • keep_values：如果提供的指标值是一个空值或者缺失值则跳过，不参与聚合计算；否则使用该指标值参与聚合计算。
• format：指定数值的输出格式，比如 #、#0.00。默认 null。
• lag：间隔多少个桶统计一次差异。默认 1。必须是一个正整数。

1、统计每天订单的消费总额以及每周的变化情况。

GET kibana_sample_data_ecommerce/_search
{
  "size": 0,
  "track_total_hits": true,
  "aggs": {
    "date_histogram_order_date": {
      "date_histogram": {
        "field": "order_date",
        "calendar_interval": "day",
        "format": "yyyy-MM-dd"
      },
      "aggs": {
        "sum_taxful_total_price": {
          "sum": {
            "field": "taxful_total_price"
          }
        },
        "serial_diff_sum_taxful_total_price": {
          "serial_diff": {
            "buckets_path": "sum_taxful_total_price",
            "lag": 7
          }
        }
      }
    }
  }
}

截取部分的聚合结果如下：文章来源地址https://www.toymoban.com/news/detail-739121.html

{
  "key_as_string" : "2022-07-14",
  "key" : 1657756800000,
  "doc_count" : 146,
  "sum_taxful_total_price" : {
    "value" : 10578.53125
  }
},
......（省略中间的聚合结果）
{
  "key_as_string" : "2022-07-21",
  "key" : 1658361600000,
  "doc_count" : 152,
  "sum_taxful_total_price" : {
    "value" : 11185.3671875
  },
  "serial_diff_sum_taxful_total_price" : {
    "value" : 606.8359375
  }
}

到了这里，关于Elasticsearch使用篇 - 管道聚合的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！