Druid: Rollups with First and Last Aggregations

First written October 2024, last updated September 2026.

Since Apache Druid 29, first and last aggregations for numerical values can be performed at ingestion. This expands the use cases of roll-ups. We’ll showcase an example and provide insights into additional use cases for Druid rollups.

What is a Rollup?

A rollup is a pre-aggregation done at ingestion time, which can dramatically reduce storage requirements for a Druid datasource (table). The trade-off is sacrificing the ability to query individual rows and losing precise timestamp granularity. Refer to the official documentation on rollup for an in-depth overview.

Rollup is a pre-aggregation, and first and last are aggregation functions. Prior to Apache Druid 29, first and last pre-aggregation only supported string data types. Since version 29, first and last also support numerical values for pre-aggregation, leading to more use cases for rollups and enabling quick data granularity reduction.

Here are some use cases for the first and last aggregation functions:

• Tracking opening and closing stock market prices.
• Creating discrete sensor readings for reports, such as hourly temperature readings.
• Creating session tracking activity for users accessing a website.

Defining a first and last Ingestion Rollup in Apache Druid

To create a first or last pre-aggregation, use the numerical data type with a First or Last suffix:

{  "type": "doubleLast",  "name": "latitude",  "fieldName": "latitude"}

{  "type": "longFirst",  "name": "speed",  "fieldName": "speed"}

Under the Hood

When pre-aggregations of first and last are computed, the timestamp of the aggregation is also stored. This ensures that when query-time aggregation is performed, the correct value is used. Query the column without the aggregation function and the resulting value appears as JSON with an lhs element for the actual timestamp of the value and rhs for the actual value.

SELECT "latitude" FROM opensky LIMIT 1

{"lhs":1728344664000,"rhs":48.2982}

The actual datatype is COMPLEX<serializablePairLongDouble>. Check out the package org.apache.druid.query.aggregation to fully appreciate this feature. While it is displayed as JSON, it isn't JSON (JSON functions, such as json_query(), don't operate on them).

Prior to Apache Druid 29, only string data types were handled this way; numerical values preserved only the value (the rhs), resulting in scenarios where the actual first or last aggregations were ambiguous.

Use Case: Sub-sampling Historical Data

For geospatial data, real-time information on all data points is vital for current events, so rollups may seem unsuitable. However, over time, keeping all those data points becomes less important. Using Druid rollups with the first or last aggregate provides an elegant way to sub-sample historical data, reducing storage costs.

At ingestion time, set the queryGranularity to none, while also enabling rollups.

   "metricsSpec": [    {    "type": "doubleLast",    "name": "latitude",    "fieldName": "latitude"    },    {    "type": "doubleLast",    "name": "longitude",    "fieldName": "longitude"    }    ...  ],  "granularitySpec": {    "type": "uniform",    "segmentGranularity": "fifteen_minute",    "queryGranularity": "none",    "rollup": true,    "intervals": null  }

To track aircraft activity, query the dataset as though it's an aggregation. For data that has not been rolled up, it acts as if it were queried without aggregation.

SELECT       __time,       id,       callsign,       latest("latitude") "latitude",       latest("longitude") "longitude",       latest("velocity") "velocity"  FROM opensky GROUP BY 1, 2, 3 ORDER BY __time ASC

The benefit is realized when the need for all data points diminishes and Druid compaction is used to quickly purge data.

Compaction

Compaction is performed by running a compaction task. The primary components are the datasource, time interval, and granularity specification. In this example, we increase segment granularity from 15 minutes to 1 hour and, more importantly, set the query granularity to 5 minutes.

{  "type" : "compact",  "dataSource" : "opensky",  "interval": "2024-10-09T12:00:00.000Z/2024-10-09T14:00:00.000Z",  "tuningConfig" : {    "type" : "index_parallel",    "maxRowsPerSegment" : 500000,    "maxRowsInMemory" : 25000  },  "granularitySpec": {    "type": "uniform",    "segmentGranularity": "hour",    "queryGranularity": "five_minute",    "rollup": true  }}

With this single administration task (no coding necessary), data is reduced once the compaction is completed. Because the last aggregation is performed on all metrics (latitude, longitude, position, velocity, altitude, etc.), the last data point for a given aircraft is the one that remains in this datasource.

Using the same query pulls in the aggregated data for the completed historical data, with no query changes necessary.

Storage Savings (How Effective Are the Rollups)

It’s always good to verify how well rollups work.

SELECT       'opensky'                         "Datasource",       SUM("count")                      "Logical Count",       COUNT("count")                    "Physical Count",       SUM("count")/(COUNT("count")*1.0) "Rollup Factor"  FROM opensky GROUP BY 1

Before Compaction

Before compaction, exact matches can lead to rollups when source systems provide duplicate events, such as opensky.

After Compaction

With timestamps rounded to the query granularity and matching dimensions, a rollup occurs.

Query

The query is written as an aggregation, so it handles rolled-up and raw data alike. If the rollup is perfect, then no additional aggregation is done, other than the single aggregated row. However, if the rollup is imperfect, that aggregation is then completed by this query.

SELECT       __time,       id,       callsign,       latest("latitude") "latitude",       latest("longitude") "longitude",       latest("velocity") "velocity",       sum("count") "count"  FROM opensky GROUP BY 1, 2, 3 ORDER BY __time ASC LIMIT 5

The count metric tracks how many records were “rolled up”.

Geospatial Data Types

Geospatial data types (geo and spatial) aren't supported by these aggregations, which is why this example uses latitude and longitude as double metrics.

Takeaways

• Check out the GitHub dev-local-demos repository for an example used in this article. It uses a container-based toolkit provided at dev-local. This repository also includes demonstrations like druid-late.
• If you are running on Apple silicon, check out the druid-m1 project for building an arm64-based image. The project has been updated to use version 30.0.1 of Apache Druid.
• With first and last support for numerical data, there are more opportunities to use rollups within Druid, as illustrated in this article. The inclusion of the timestamp in the internal complex data type ensures the last value is truly the last value (and the first value is the first value).
• Rollups are a compromise of precision for storage, and the inclusion of a given dimension or metric can dramatically reduce rollup effectiveness. Verify the effectiveness of your rollup after any datasource configuration updates.

If you're storing more resolution than anyone queries, talk to us.