当前位置：首页 > news >正文

第三阶段 27 · pipeline 管道聚合（同比/环比/累计/移动平均）

news 2026/8/3 17:00:24

阶段：第三阶段补充 / 聚合能力
ES：pipeline aggregation | PostgreSQL：窗口函数（SUM() OVER、LAG()）

1. 概念

前面 20–24 篇的聚合都是对文档算（指标、分桶）。
管道聚合（pipeline agg）不碰文档，而是拿「其它聚合的结果」再算一层——
比如对每月的销售额桶，算累计值、环比增长、移动平均。

两类：

类型	作用	典型
`parent`（父管道）	在同级桶序列上算新指标，结果写回每个桶	`cumulative_sum`、`derivative`、`moving_fn`
`sibling`（兄弟管道）	对整组桶汇总出一个值	`max_bucket`、`avg_bucket`、`sum_bucket`

关键：管道聚合用buckets_path指向要引用的聚合结果（像“公式引用单元格”）。

2. PostgreSQL 对照

-- 每月销售额 + 累计 + 环比（窗口函数）SELECTmonth,SUM(amount)ASmonthly,SUM(SUM(amount))OVER(ORDERBYmonth)AScumulative,-- cumulative_sumSUM(amount)-LAG(SUM(amount))OVER(ORDERBYmonth)ASmom-- derivativeFROMsalesGROUPBYmonthORDERBYmonth;

ES 的管道聚合就是 ES 版「对分组结果再套窗口函数」。

3. ES DSL

3.1 累计求和 + 环比（parent 管道）

GET sales_idx/_search { "size": 0, "aggs": { "by_month": { "date_histogram": { "field": "invoice_dt", "calendar_interval": "month" }, "aggs": { "monthly": { "sum": { "field": "amount" } }, "cumulative": { "cumulative_sum": { "buckets_path": "monthly" } // 累计 }, "mom": { "derivative": { "buckets_path": "monthly" } // 环比差值 }, "moving_avg_3": { "moving_fn": { // 3 期移动平均 "buckets_path": "monthly", "window": 3, "script": "MovingFunctions.unweightedAvg(values)" } } } } } }

3.2 找销售额最高的月份（sibling 管道）

GET sales_idx/_search { "size": 0, "aggs": { "by_month": { "date_histogram": { "field": "invoice_dt", "calendar_interval": "month" }, "aggs": { "monthly": { "sum": { "field": "amount" } } } }, "best_month": { "max_bucket": { "buckets_path": "by_month>monthly" } // > 表示钻进子聚合 } } }

buckets_path语法：聚合名>子聚合名；>是“进入下一层桶”，类似路径分隔符。

4. Spring Boot 实现

@ComponentpublicclassDoc27PipelineAgg{@AutowiredprivateElasticsearchClientelasticsearchClient;/** 每月销售额 + 累计值 */publicMap<String,Double[]>monthlyWithCumulative(StringindexName)throwsIOException{SearchResponse<Void>resp=elasticsearchClient.search(s->s.index(indexName).size(0).aggregations("by_month",a->a.dateHistogram(dh->dh.field("invoice_dt").calendarInterval(CalendarInterval.Month)).aggregations("monthly",m->m.sum(su->su.field("amount"))).aggregations("cumulative",c->c.cumulativeSum(cs->cs.bucketsPath(bp->bp.single("monthly"))))),Void.class);Map<String,Double[]>out=newLinkedHashMap<>();for(DateHistogramBucketb:resp.aggregations().get("by_month").dateHistogram().buckets().array()){doublemonthly=b.aggregations().get("monthly").sum().value();// 管道聚合结果也是一个 simpleValuedoublecumulative=b.aggregations().get("cumulative").simpleValue().value();out.put(b.keyAsString(),newDouble[]{monthly,cumulative});}returnout;}}

import：...aggregations.CalendarInterval、...aggregations.DateHistogramBucket。
管道聚合结果读取用.simpleValue().value()（cumulative_sum/derivative/moving_fn都是）。
buckets_path在客户端用BucketsPath：单路径bp.single("monthly")。