06c — Pipeline aggregations

Pipeline aggregations operate on the output of other aggregations — not on raw documents. They let you compute derivatives, running totals, moving averages, and per-bucket ratios, and filter buckets by metric values. This part also covers the practical pitfalls that trip teams in production.

Goals for this part

Understand the two pipeline families: parent and sibling
Use derivative, cumulative_sum, moving_fn, bucket_script, and bucket_selector
Use sibling aggregations (max_bucket, avg_bucket) across a series
Avoid the most common aggregation performance traps

6.4 — Pipeline aggregations

Pipeline aggregations operate on the output of other aggregations — their input is bucket values or metrics from a sibling/parent agg, not raw documents. They fall into two families:

Family	Examples	Input
Parent	`derivative`, `cumulative_sum`, `moving_fn`, `bucket_script`, `bucket_selector`	Each bucket in the parent series
Sibling	`avg_bucket`, `max_bucket`, `min_bucket`, `sum_bucket`, `stats_bucket`	All buckets of a sibling agg

6.4.1 `buckets_path` syntax

Pipeline aggs reference their inputs with buckets_path. The path walks the agg tree using > as a separator:

"monthly_revenue>_count"          — the doc_count of the monthly_revenue agg
"monthly_revenue>avg_order_value" — the avg_order_value sub-agg inside each bucket

6.4.2 `gap_policy`

When a bucket has no data (e.g., a month with zero orders), pipeline aggs face a gap. gap_policy controls this:

skip — omit the bucket from the pipeline output
insert_zeros — treat the gap as zero (use for cumulative sums)

6.4.3 Full example: monthly revenue analytics

GET shop-orders-demo/_search
{
  "size": 0,
  "query": {
    "range": { "created_at": { "gte": "2024-01-01", "lt": "2025-01-01" } }
  },
  "aggs": {
    "monthly": {
      "date_histogram": {
        "field": "created_at",
        "calendar_interval": "month",
        "min_doc_count": 0,
        "extended_bounds": { "min": "2024-01-01", "max": "2024-12-31" }
      },
      "aggs": {
        "revenue":          { "sum": { "field": "total_amount"    } },
        "discount_total":   { "sum": { "field": "discount_amount" } },
        "subtotal_total":   { "sum": { "field": "subtotal_amount" } },

        "mom_derivative": {
          "derivative": {
            "buckets_path": "revenue",
            "gap_policy": "skip"
          }
        },

        "cumulative_revenue": {
          "cumulative_sum": {
            "buckets_path": "revenue",
            "gap_policy": "insert_zeros"
          }
        },

        "moving_avg_3m": {
          "moving_fn": {
            "buckets_path": "revenue",
            "window": 3,
            "script": "MovingFunctions.unweightedAvg(values)",
            "gap_policy": "skip"
          }
        },

        "discount_ratio": {
          "bucket_script": {
            "buckets_path": {
              "disc": "discount_total",
              "sub":  "subtotal_total"
            },
            "script": "params.sub > 0 ? params.disc / params.sub : 0"
          }
        }
      }
    },

    "best_month": {
      "max_bucket": {
        "buckets_path": "monthly>revenue"
      }
    },

    "avg_monthly_revenue": {
      "avg_bucket": {
        "buckets_path": "monthly>revenue"
      }
    }
  }
}

What each pipeline does:

Agg	Type	What it produces
`mom_derivative`	Parent	Month-over-month revenue change (positive = growth)
`cumulative_revenue`	Parent	Running total of revenue from Jan → Dec
`moving_avg_3m`	Parent	3-month unweighted moving average
`discount_ratio`	Parent	`discount / subtotal` per month — a computed ratio
`best_month`	Sibling	The single month with peak revenue, across all buckets
`avg_monthly_revenue`	Sibling	Average monthly revenue across all 12 months

moving_avg is deprecated. The dedicated moving_avg aggregation was deprecated in Elasticsearch 6.4. Use moving_fn with MovingFunctions.unweightedAvg(values), MovingFunctions.ewma(values, alpha), or MovingFunctions.holt(values, alpha, beta) instead.

6.4.4 `bucket_selector` — filtering buckets by metric value

Remove buckets where the average order value is below a threshold:

"aggs": {
  "by_country": {
    "terms": { "field": "delivery.country_code", "size": 20 },
    "aggs": {
      "avg_value": { "avg": { "field": "total_amount" } },
      "only_high_value_markets": {
        "bucket_selector": {
          "buckets_path": { "avgVal": "avg_value" },
          "script": "params.avgVal >= 150"
        }
      }
    }
  }
}

Buckets where avg_value < 150 are silently dropped from the response. This is a parent pipeline agg — it runs after all bucket metrics are computed.

6.5 — Practical notes and pitfalls

High-cardinality `terms` — don’t do it

terms on order_id, customer.id, or any UUID-like field will: - Force Elasticsearch to build a massive in-memory bucket structure per shard - Produce inaccurate results because shard_size can never be large enough - Potentially cause OOM errors in extreme cases

Use composite aggregation instead when you need to enumerate all values (pagination of buckets):

"aggs": {
  "paginated_customers": {
    "composite": {
      "size": 1000,
      "sources": [
        { "customer": { "terms": { "field": "customer.id" } } }
      ],
      "after": { "customer": "last-seen-customer-id" }
    }
  }
}

Pass the after_key from the previous response as after in the next request to page through all buckets.

Always scope with a time/range filter

Unbounded aggregations across the full index are expensive and rarely what you want. Always apply a query with a range on created_at (or another date field) before running analytics aggregations.

`size: 0` is not optional for analytics

Fetching hits alongside aggregations doubles serialisation cost. Unless you genuinely need the documents, set "size": 0.

Memory and accuracy trade-offs — summary

Agg	Accuracy	Memory lever
`terms`	Approximate (distributed)	`shard_size`
`cardinality`	Approximate (HLL++)	`precision_threshold`
`percentiles`	Approximate (TDigest)	`compression`
`date_histogram`, `range`, `filter`	Exact	N/A
Pipeline aggs	Exact (operate on already-computed values)	N/A

Lab — Pipeline aggregations and bucket filtering

Part A: Monthly sales trend with pipeline analytics

Goal: Build a full time-series analytics query with multiple pipeline aggs.

Step 1 — Monthly revenue for the data’s year range with min_doc_count: 0 and extended_bounds. Add derivative (month-over-month change), cumulative_sum, and moving_fn (3-month average). Use a sibling max_bucket to identify the peak month:

GET /shop-orders-demo/_search
{
  "size": 0,
  "aggs": {
    "monthly": {
      "date_histogram": {
        "field": "created_at",
        "calendar_interval": "month",
        "min_doc_count": 0
      },
      "aggs": {
        "revenue": { "sum": { "field": "total_amount" } },

        "mom_growth": {
          "derivative": {
            "buckets_path": "revenue",
            "gap_policy": "skip"
          }
        },

        "running_total": {
          "cumulative_sum": {
            "buckets_path": "revenue",
            "gap_policy": "insert_zeros"
          }
        },

        "smoothed_3m": {
          "moving_fn": {
            "buckets_path": "revenue",
            "window": 3,
            "script": "MovingFunctions.unweightedAvg(values)",
            "gap_policy": "skip"
          }
        }
      }
    },

    "peak_month": {
      "max_bucket": {
        "buckets_path": "monthly>revenue"
      }
    },

    "avg_monthly": {
      "avg_bucket": {
        "buckets_path": "monthly>revenue"
      }
    }
  }
}

Questions: 1. Which month had the highest revenue (peak_month)? 2. In which months did mom_growth turn negative (revenue declined vs previous month)? 3. What does the smoothed_3m value tell you that raw monthly revenue doesn’t? 4. What happens to mom_growth in the first bucket? Why is there no value?

Step 2 — Remove min_doc_count: 0 and observe what happens to the cumulative sum when a month has no data. Re-add it and verify all months are present.

Part B: Computed ratios and bucket filtering

Goal: Use bucket_script to compute a per-bucket metric and bucket_selector to filter results.

Step 3 — Bucket by delivery.country_code, compute sum of total_amount and sum of discount_amount. Use bucket_script to compute the discount rate per country:

GET /shop-orders-demo/_search
{
  "size": 0,
  "aggs": {
    "by_country": {
      "terms": { "field": "delivery.country_code", "size": 20 },
      "aggs": {
        "total_revenue":   { "sum": { "field": "total_amount"    } },
        "total_discounts": { "sum": { "field": "discount_amount" } },
        "avg_value":       { "avg": { "field": "total_amount"    } },

        "discount_rate": {
          "bucket_script": {
            "buckets_path": {
              "rev":  "total_revenue",
              "disc": "total_discounts"
            },
            "script": "params.rev > 0 ? params.disc / params.rev : 0"
          }
        }
      }
    }
  }
}

Questions: 1. Which country has the highest discount rate? 2. Is there a correlation between discount rate and average order value?

Step 4 — Add a bucket_selector to keep only countries where avg_value >= 120. Compare the bucket count before and after:

GET /shop-orders-demo/_search
{
  "size": 0,
  "aggs": {
    "by_country": {
      "terms": { "field": "delivery.country_code", "size": 20 },
      "aggs": {
        "avg_value": { "avg": { "field": "total_amount" } },
        "only_premium_markets": {
          "bucket_selector": {
            "buckets_path": { "avgVal": "avg_value" },
            "script": "params.avgVal >= 120"
          }
        }
      }
    }
  }
}

Questions: 1. How many countries remain after the bucket_selector filter? 2. bucket_selector is a “parent pipeline agg” — what does this mean in terms of execution order?

Step 5 — Deliberately try running terms on a high-cardinality field (order_id) and observe the warning or performance impact. Then use composite as the correct alternative to iterate through a subset:

GET /shop-orders-demo/_search
{
  "size": 0,
  "aggs": {
    "paginated_orders": {
      "composite": {
        "size": 10,
        "sources": [
          { "order": { "terms": { "field": "order_id" } } }
        ]
      }
    }
  }
}

Take the after_key from the response and fetch the next 10:

GET /shop-orders-demo/_search
{
  "size": 0,
  "aggs": {
    "paginated_orders": {
      "composite": {
        "size": 10,
        "sources": [
          { "order": { "terms": { "field": "order_id" } } }
        ],
        "after": { "order": "<paste after_key value here>" }
      }
    }
  }
}

Takeaway: Parent pipeline aggs (derivative, cumulative_sum, moving_fn, bucket_script, bucket_selector) run after all bucket metrics are computed. Sibling aggs (max_bucket, avg_bucket) compare across the entire bucket set. Both require min_doc_count: 0 on the parent histogram for a complete series. Use composite instead of high-cardinality terms whenever you need to enumerate all values.