拓冰网站定制 
阶段：第三阶段补充 / 聚合能力
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")。
5. 坑与最佳实践
- 管道聚合依赖“有序桶序列”：多和
date_histogram/histogram搭配，桶要按序。 derivative首个桶没有环比值（没有前一项），前端要容错。buckets_path写错最常见：层级用>，名字要和上面的聚合名完全一致。gap_policy：桶里缺值（某月无数据）时用skip/insert_zeros控制行为，避免断链。moving_avg已废弃，用moving_fn+ Painless（如MovingFunctions.unweightedAvg）。
下一篇
30-index-document-写入.md（进入第四阶段：写入与索引管理）。