阶段第三阶段补充 / 聚合能力ESpipeline aggregation | PostgreSQL窗口函数SUM() OVER、LAG()1. 概念前面 20–24 篇的聚合都是对文档算指标、分桶。管道聚合pipeline agg不碰文档而是拿「其它聚合的结果」再算一层——比如对每月的销售额桶算累计值、环比增长、移动平均。两类类型作用典型parent父管道在同级桶序列上算新指标结果写回每个桶cumulative_sum、derivative、moving_fnsibling兄弟管道对整组桶汇总出一个值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 DSL3.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_monthmonthly } // 表示钻进子聚合 } } }buckets_path语法聚合名子聚合名是“进入下一层桶”类似路径分隔符。4. Spring Boot 实现ComponentpublicclassDoc27PipelineAgg{AutowiredprivateElasticsearchClientelasticsearchClient;/** 每月销售额 累计值 */publicMapString,Double[]monthlyWithCumulative(StringindexName)throwsIOException{SearchResponseVoidrespelasticsearchClient.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);MapString,Double[]outnewLinkedHashMap();for(DateHistogramBucketb:resp.aggregations().get(by_month).dateHistogram().buckets().array()){doublemonthlyb.aggregations().get(monthly).sum().value();// 管道聚合结果也是一个 simpleValuedoublecumulativeb.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进入第四阶段写入与索引管理。