4. Pipeline聚合实战：derivative、moving_avg、cumulative_sum、bucket_script、bucket_selector

Pipeline聚合，说白了就是「聚合的聚合」。

普通聚合是对原始数据做统计，而Pipeline聚合是对聚合结果再做一次运算。我刚开始接触这个概念时，也觉得有点绕。但用过几次后，你会发现它特别适合做时序数据的趋势分析、环比计算、数据筛选这类场景。

这一章，我们重点讲五个常用的Pipeline聚合：derivative、moving_avg、cumulative_sum、bucket_script、bucket_selector。每个我都会结合实战案例来讲。

4.1 derivative：计算相邻桶的差值

derivative 聚合，用来计算相邻两个桶之间的差值。说白了，就是「这小时比上小时多了多少」或者「今天比昨天涨了多少」。

我在项目中遇到过这样一个需求：监控线上服务的QPS，想看每5分钟的增量变化。如果QPS突然飙升，说明可能有异常流量。用 derivative 就很方便。

核心思路：先做date_histogram按时间分桶，再对每个桶内的QPS求平均值，最后用derivative计算相邻桶的差值。

看个例子：

GET /logs/_search
{
  "size": 0,
  "aggs": {
    "per_minute": {
      "date_histogram": {
        "field": "@timestamp",
        "interval": "minute"
      },
      "aggs": {
        "avg_qps": {
          "avg": {
            "field": "qps"
          }
        },
        "qps_derivative": {
          "derivative": {
            "buckets_path": "avg_qps"
          }
        }
      }
    }
  }
}

返回结果里，每个桶会多一个 qps_derivative 字段，值就是当前桶的 avg_qps 减去上一个桶的 avg_qps。

小提示：第一个桶没有上一个桶，所以它的derivative值是null。这在做可视化时要注意处理一下。

4.2 moving_avg：平滑波动，看清趋势

moving_avg 就是移动平均。它把前面N个桶的值取平均，用来平滑数据中的毛刺。

为什么要用移动平均？你想想看，原始数据可能忽高忽低，比如服务器CPU使用率，一会儿30%，一会儿80%，根本看不出趋势。但用移动平均一平滑，趋势就清晰了。

我个人习惯在监控大盘里用 moving_avg，配合 derivative 一起看。一个看增量，一个看趋势，效果很好。

GET /metrics/_search
{
  "size": 0,
  "aggs": {
    "per_hour": {
      "date_histogram": {
        "field": "timestamp",
        "interval": "hour"
      },
      "aggs": {
        "avg_cpu": {
          "avg": {
            "field": "cpu_usage"
          }
        },
        "cpu_moving_avg": {
          "moving_avg": {
            "buckets_path": "avg_cpu",
            "window": 5,
            "model": "simple"
          }
        }
      }
    }
  }
}

这里 window: 5 表示取前5个桶的平均值。model 参数可以选 simple（简单平均）、linear（线性加权）、ewma（指数加权）等。我一般用 simple 就够了，除非数据波动特别大才考虑 ewma。

注意：window值越大，平滑效果越强，但延迟也越大。如果window设成10，那前9个桶的moving_avg都是null。我曾经踩过这个坑，可视化时发现前面一大段空白，排查了半天才发现是window设太大了。

4.3 cumulative_sum：累加求和

cumulative_sum 就是累加。从第一个桶开始，把当前桶及之前所有桶的值加起来。

这个聚合特别适合做「累计增长」的场景。比如统计网站每天的注册用户数，然后看累计注册总量。或者统计每个月的销售额，看年度累计销售额。

GET /sales/_search
{
  "size": 0,
  "aggs": {
    "per_month": {
      "date_histogram": {
        "field": "order_date",
        "interval": "month",
        "format": "yyyy-MM"
      },
      "aggs": {
        "monthly_sales": {
          "sum": {
            "field": "amount"
          }
        },
        "cumulative_sales": {
          "cumulative_sum": {
            "buckets_path": "monthly_sales"
          }
        }
      }
    }
  }
}

结果里，cumulative_sales 会逐月累加。1月是1月的销售额，2月是1月+2月的和，3月是1月+2月+3月的和，以此类推。

小技巧：cumulative_sum可以和derivative配合使用。比如先算累计值，再算累计值的增量，就能看出「每天新增了多少累计量」。我在做用户增长分析时经常这么用。

4.4 bucket_script：自定义计算

bucket_script 是Pipeline聚合里最灵活的一个。它允许你用脚本对多个聚合结果做自定义运算。

说白了，就是给你一个「计算器」，你可以把几个聚合结果拿过来，加减乘除随便算。

我记得有一次，业务方想看「订单转化率」。需要先统计「访问次数」和「下单次数」，然后用下单次数除以访问次数。用 bucket_script 一行脚本就搞定了。

GET /events/_search
{
  "size": 0,
  "aggs": {
    "per_day": {
      "date_histogram": {
        "field": "event_date",
        "interval": "day"
      },
      "aggs": {
        "visits": {
          "filter": { "term": { "event_type": "visit" } }
        },
        "orders": {
          "filter": { "term": { "event_type": "order" } }
        },
        "conversion_rate": {
          "bucket_script": {
            "buckets_path": {
              "visit_count": "visits._count",
              "order_count": "orders._count"
            },
            "script": "params.order_count / params.visit_count * 100"
          }
        }
      }
    }
  }
}

这里 buckets_path 定义了输入变量，script 里写计算逻辑。支持Painless脚本语法，加减乘除、条件判断都可以。

注意：bucket_script的脚本里，变量名要用 params. 前缀引用。比如上面例子里的 params.visit_count。我第一次写时忘了加 params.，结果报错半天没找到原因。

4.5 bucket_selector：按条件过滤桶

bucket_selector 用来过滤掉不符合条件的桶。它和 bucket_script 很像，也是写脚本做判断，但它的作用是「保留或丢弃整个桶」。

为什么要过滤桶？举个例子：你统计了每个小时的错误日志数量，但只想看「错误数超过100的时段」。这时候 bucket_selector 就派上用场了。

GET /logs/_search
{
  "size": 0,
  "aggs": {
    "per_hour": {
      "date_histogram": {
        "field": "@timestamp",
        "interval": "hour"
      },
      "aggs": {
        "error_count": {
          "filter": { "term": { "level": "ERROR" } }
        },
        "high_error_hours": {
          "bucket_selector": {
            "buckets_path": {
              "count": "error_count._count"
            },
            "script": "params.count > 100"
          }
        }
      }
    }
  }
}

结果里，只有错误数大于100的桶会被保留。其他桶直接消失。

踩坑提醒：bucket_selector是在聚合计算完成后才执行的过滤。它不会减少计算量，只是减少返回结果的数据量。我曾经以为它能提升性能，后来发现该算的桶一个都没少算。

4.6 实战组合：一个完整的监控分析案例

最后，我把这几个Pipeline聚合组合起来，做一个完整的实战案例。

假设我们要分析一个电商网站的访问数据，需求是：

  1. 按小时统计访问量
  2. 计算每小时的访问量增量（derivative）
  3. 计算3小时移动平均（moving_avg）
  4. 计算累计访问量（cumulative_sum）
  5. 计算转化率（bucket_script：下单数/访问数）
  6. 只保留访问量大于1000的时段（bucket_selector）
GET /ecommerce/_search
{
  "size": 0,
  "aggs": {
    "per_hour": {
      "date_histogram": {
        "field": "timestamp",
        "interval": "hour"
      },
      "aggs": {
        "visits": {
          "value_count": { "field": "user_id" }
        },
        "orders": {
          "filter": { "term": { "action": "order" } }
        },
        "visit_derivative": {
          "derivative": { "buckets_path": "visits" }
        },
        "visit_moving_avg": {
          "moving_avg": {
            "buckets_path": "visits",
            "window": 3
          }
        },
        "visit_cumulative": {
          "cumulative_sum": { "buckets_path": "visits" }
        },
        "conversion_rate": {
          "bucket_script": {
            "buckets_path": {
              "v": "visits",
              "o": "orders._count"
            },
            "script": "params.o / params.v * 100"
          }
        },
        "high_traffic_hours": {
          "bucket_selector": {
            "buckets_path": { "v": "visits" },
            "script": "params.v > 1000"
          }
        }
      }
    }
  }
}

这个查询跑下来，你就能拿到一份非常完整的时序分析数据。每个小时都有访问量、增量、移动平均、累计值、转化率，而且只保留高流量时段。

我的建议：在实际项目中，Pipeline聚合不要嵌套太深。超过3层嵌套，可读性会急剧下降，排查问题也困难。我一般控制在2层以内，如果逻辑复杂，就拆成多个查询，或者用Kibana的Vega做二次处理。

好了，Pipeline聚合的五个核心成员就讲完了。它们各有各的用途，但组合起来威力更大。下一章我们聊聊如何把这些聚合结果做成可视化报表，让数据真正「说话」。

← 上一章 📖 返回目录 下一章 →