Elasticsearch Pipeline Aggregations: Practical Examples

Elasticsearch 2.0.0 beta introduced a wide range of performance improvements, but one of its most anticipated new features was pipeline aggregations.

Pipeline aggregations make it possible to perform calculations on the results of other Elasticsearch aggregations. Instead of working directly with documents, they use the buckets and metrics produced by previous aggregations as their input.

In this guide, we look at how Elasticsearch pipeline aggregations work and explore practical examples of average, maximum, minimum, sum, derivative, cumulative sum, bucket script and bucket selector aggregations.

What are Elasticsearch pipeline aggregations?

Elasticsearch pipeline aggregations can broadly be divided into two types:

• Parent pipeline aggregations – A pipeline aggregation calculates a result and adds that result to the buckets or aggregations produced by its parent aggregation.
• Sibling pipeline aggregations – An existing aggregation becomes the input for a pipeline aggregation, which produces a new aggregation at the same level as the original aggregation.

The examples below demonstrate both types and show how they can be used to perform additional calculations on aggregated Elasticsearch data.

Pipeline aggregations do not support sub-aggregations themselves, but they can be chained. When several pipeline aggregations are chained together, the final response contains the output of each aggregation in the chain.

Using buckets_path in Elasticsearch pipeline aggregations

To tell a pipeline aggregation which previous aggregation it should use as input, Elasticsearch uses the buckets_path parameter.

The basic syntax is:

"buckets_path": "Aggregation>Metric"

Here, buckets_path refers to an aggregation and a metric within that aggregation.

We will use this throughout the examples below.

Setting up the Elasticsearch sample data

First, let us create an index using data based on the Elasticsearch Definitive Guide.

For these examples, I used the Sense Chrome extension. At the time this article was written, Elasticsearch 2.0.0 beta did not support installation of the Marvel plugin from the command prompt.

POST /cars/transactions/_bulk
{ "index": {}}
{ "price" : 10000, "color" : "red", "make" : "honda", "sold" : "2014-10-28" }
{ "index": {}}
{ "price" : 20000, "color" : "red", "make" : "honda", "sold" : "2014-11-05" }
{ "index": {}}
{ "price" : 30000, "color" : "green", "make" : "ford", "sold" : "2014-05-18" }
{ "index": {}}
{ "price" : 15000, "color" : "blue", "make" : "toyota", "sold" : "2014-07-02" }

For brevity, only four documents are shown here. In total, I inserted 16 records using the same schema.

Sample data: Elasticsearch pipeline aggregation sample data on GitHub.

Sibling pipeline aggregations in Elasticsearch

Average bucket aggregation: avg_bucket

The avg_bucket aggregation is a sibling pipeline aggregation. It calculates the average value of a specified metric from another sibling aggregation.

The sibling aggregation needs to be a multi-bucket aggregation. In this example, cars are grouped by the month in which they were sold. We first calculate the total sales for each month and then use a pipeline aggregation to calculate the average monthly sales.

GET /cars/transactions/_search?search_type=count
{
   "aggs":{
      "sales_per_month":{
         "date_histogram":{
            "field":"sold",
            "interval":"month",
            "format":"yyyy-MM-dd"
         },
         "aggs":{
            "monthly_sum":{
               "sum":{
                  "field":"price"
               }
            }
         }
      },
      "avg_monthly_sales":{
         "avg_bucket":{
            "buckets_path":"sales_per_month>monthly_sum"
         }
      }
   }
}

The key part of this query is:

"buckets_path": "sales_per_month>monthly_sum"

Here, sales_per_month identifies the aggregation and monthly_sum identifies the metric that should be used by the pipeline aggregation.

The sales_per_month aggregation gives us the total value of cars sold each month. The sibling aggregation avg_monthly_sales then calculates the average of those monthly totals.

"aggregations": {
      "sales_per_month": {
         "buckets": [
            {
               "key_as_string": "2014-01-01",
               "key": 1388534400000,
               "doc_count": 3,
               "monthly_sum": {
                  "value": 185000
               }
            },
            {
               "key_as_string": "2014-02-01",
               "key": 1391212800000,
               "doc_count": 1,
               "monthly_sum": {
                  "value": 25000
               }
            },
            ……………… 14 more records
         ]
      },
      "avg_monthly_sales": {
         "value": 41900
      }
   }

The resulting avg_monthly_sales aggregation appears alongside sales_per_month, rather than inside one of its buckets. That makes it a sibling pipeline aggregation.

We can also change the interval from month to quarter to calculate the average quarterly sales total.

"avg_quaterly_sales": {
   "value": 104750
}

Maximum and minimum bucket aggregations: max_bucket and min_bucket

Like the average bucket aggregation, max_bucket and min_bucket are sibling pipeline aggregations.

They produce an aggregation alongside the input aggregation and make it straightforward to identify the bucket containing the highest or lowest value for a particular metric.

Using our previous example, we can replace avg_monthly_sales with:

"max_monthly_sales": {
   "max_bucket": {
      "buckets_path": "sales_per_month>monthly_sum"
   }
}

To find the minimum monthly sales value, use:

"min_monthly_sales": {
   "min_bucket": {
      "buckets_path": "sales_per_month>monthly_sum"
   }
}

The minimum result is:

"min_monthly_sales": {
   "value": 10000,
   "keys": [
      "2014-10-01"
   ]
}

And the maximum result is:

"max_monthly_sales": {
   "value": 185000,
   "keys": [
      "2014-01-01"
   ]
}

This gives us both the minimum or maximum value and the corresponding bucket key.

Sum bucket aggregation: sum_bucket

The sum_bucket aggregation is another sibling pipeline aggregation. It calculates the sum of a metric across multiple buckets.

For example, we can first filter the original aggregation so that it only includes BMW cars:

"query" : {
   "match" : {
      "make" : "bmw"
   }
},
"aggs" …..
……

We can then calculate monthly BMW sales and use a sum_bucket aggregation to calculate the total:

"sum_bmw_sales": {
   "sum_bucket": {
      "buckets_path": "sales_per_month>monthly_sum"
   }
}

We now have both the total BMW sales for each month and the total yearly sales for BMW.

The same approach could also be used with other search criteria, such as a date range, car color or another field.

Parent pipeline aggregations in Elasticsearch

Derivative aggregation

The derivative aggregation is a parent pipeline aggregation. The calculated derivative of a specified metric becomes part of the bucket produced by the parent aggregation.

GET /cars/transactions/_search?search_type=count
{
   "aggs": {
      "sales_per_month": {
         "date_histogram": {
            "field": "sold",
            "interval": "month",
            "format": "yyyy-MM-dd"
         },
         "aggs": {
            "monthly_sum": {
               "sum": {
                  "field": "price"
               }
            },
            "sales_deriv": {
               "derivative": {
                  "buckets_path": "monthly_sum"
               }
            }
         }
      }
   }
}

In this query, we calculate the derivative of monthly_sum.

The output looks like this:

"aggregations": {
   "sales_per_month": {
      "buckets": [
         {
            "key_as_string": "2014-01-01",
            "key": 1388534400000,
            "doc_count": 3,
            "monthly_sum": {
               "value": 185000
            }
         },
         {
            "key_as_string": "2014-02-01",
            "key": 1391212800000,
            "doc_count": 1,
            "monthly_sum": {
               "value": 25000
            },
            "sales_deriv": {
               "value": -160000
            }
         },
         {
            "key_as_string": "2014-03-01",
            "key": 1393632000000,
            "doc_count": 1,
            "monthly_sum": {
               "value": 30000
            },
            "sales_deriv": {
               "value": 5000
            }
         },
         ……..13 more records
      ]
   }
}

There is no derivative value for the first bucket because calculating a derivative requires at least two data points.

Cumulative sum aggregation

The cumulative_sum aggregation is another parent pipeline aggregation. It calculates the cumulative sum of a specified metric from the input aggregation.

In our example, it can be used to calculate cumulative sales over time on a monthly basis.

We can replace the sales_deriv section from the previous query with:

"cumulative_sales": {
   "cumulative_sum": {
      "buckets_path": "monthly_sum"
   }
}

The resulting output is:

"aggregations": {
   "sales_per_month": {
      "buckets": [
         {
            "key_as_string": "2014-01-01",
            "key": 1388534400000,
            "doc_count": 3,
            "monthly_sum": {
               "value": 185000
            },
            "cumulative_sales": {
               "value": 185000
            }
         },
         {
            "key_as_string": "2014-02-01",
            "key": 1391212800000,
            "doc_count": 1,
            "monthly_sum": {
               "value": 25000
            },
            "cumulative_sales": {
               "value": 210000
            }
         },
         {
            "key_as_string": "2014-03-01",
            "key": 1393632000000,
            "doc_count": 1,
            "monthly_sum": {
               "value": 30000
            },
            "cumulative_sales": {
               "value": 240000
            }
         },
         ..13 more records..
      ]
   }
}

This aggregation makes it possible to visualize how a value accumulates over time, for example to analyze cumulative sales across peak periods or compare different products.

Bucket script aggregation

The bucket_script aggregation is a parent pipeline aggregation that uses scripts to perform arithmetic calculations on metrics within each bucket of a multi-bucket aggregation.

A common use case is calculating a percentage or performing calculations such as addition and subtraction using the results of sub-aggregations.

Suppose we want to calculate what percentage of total monthly sales comes from BMW cars. First, we need a sub-aggregation in each monthly bucket for BMW sales. We can then calculate BMW sales as a percentage of total sales.

This pipeline aggregation uses scripting. You can find more background in the Elasticsearch scripting documentation.

At the time this article was written, the example used inline scripting, which was not advised for production environments. To enable inline scripting for this example, the following setting was added to the elasticsearch.yml file:

script.inline: on

The aggregation then looks like this:

"aggs": {
   "sales_per_month": {
      "date_histogram": {
         "field": "sold",
         "interval": "month",
         "format": "yyyy-MM-dd"
      },
      "aggs": {
         "monthly_sum": {
            "sum": {
               "field": "price"
            }
         },
         "bmw_car": {
            "filter": {
               "term": {
                  "make": "bmw"
               }
            },
            "aggs": {
               "sales": {
                  "sum": {
                     "field": "price"
                  }
               }
            }
         },
         "bmw_percentage": {
            "bucket_script": {
               "buckets_path": {
                  "bmwSales": "bmw_car>sales",
                  "totalSales": "monthly_sum"
               },
               "script": "bmwSales / totalSales * 100"
            }
         }
      }
   }
}

The response is:

{
   "key_as_string": "2014-01-01",
   "key": 1388534400000,
   "doc_count": 3,
   "monthly_sum": {
      "value": 185000
   },
   "bmw_car": {
      "doc_count": 2,
      "sales": {
         "value": 160000
      }
   },
   "bmw_percentage": {
      "value": 86.48
   }
}

For January, three cars were sold. Two of them were BMWs, accounting for 86.48% of the total sales value.

Bucket selector aggregation

The bucket_selector aggregation is useful when you do not want certain buckets to appear in the output based on conditions you define.

Examples include returning only buckets where:

• a total value is greater than a particular threshold;
• a percentage exceeds a specified value;
• a count is greater than a defined number.

This is another script-based parent pipeline aggregation, so scripting needs to be enabled for this historical example.

We can add the following snippet in the same location where the sales_deriv aggregation was added:

"sales_bucket_filter": {
   "bucket_selector": {
      "buckets_path": {
         "totalSales": "monthly_sum"
      },
      "script": "totalSales >= 30000"
   }
}

The output will now contain only months where monthly sales are at least 30,000.

Elasticsearch pipeline aggregations: conclusion

Pipeline aggregations introduced many useful ways to derive additional insights from data already aggregated in Elasticsearch.

As these examples demonstrate, sibling pipeline aggregations such as avg_bucket, max_bucket, min_bucket and sum_bucket can calculate values across existing buckets, while parent pipeline aggregations such as derivative, cumulative_sum, bucket_script and bucket_selector can add additional calculations to individual buckets.

There are many real-world use cases for Elasticsearch pipeline aggregations, particularly when analyzing changes over time, totals, averages, thresholds and relationships between metrics.

Moving average aggregation is not covered here because the subject deserves a separate article.

You can find the complete set of 16 records used in these examples in the original GitHub Gist.

Archive notice: This is an older blog post about Elasticsearch 2.0.0 beta. Elasticsearch, its APIs, tooling, scripting options and recommended implementation practices have changed significantly since this article was originally published.

We keep the technical content and examples available in their original historical context for research and reference purposes rather than continuously updating them to current Elasticsearch versions. As a result, examples in this article should not be treated as current implementation guidance.

Have a question about this article or about a current implementation? Please use the contact form to get in touch.