ClickHouse: Aggregating Latest Values, Take Two

Data Cookbook Kitchen

Taking the latest value of a timeseeries is, from ClickHouse’s point of view, just a form of aggregation. In an earlier blog I showed how to set up an ingestion pipeline that continually updates the latest values to an AggregatingMergeTree table. This is a common requirement in analyzing IoT data.

But recently I encountered an interesting variant of this use case: That customer wanted to show, for each timeseries, the latest and the previous (latest-but-one) values! Let’s find out how to do this in ClickHouse!

The base data model

Let’s go with a pretty generic IoT data model. Each timeseries is identified by the tuple (device_id, parameter_id), which also forms the order key of the table. Within a timeseries, individual measurements are identified by their timestamp. The DDL is straightforward:

DROP TABLE IF EXISTS lasttwo_raw;

CREATE TABLE lasttwo_raw (
    ts              DateTime64,
    device_id       UInt64,
    parameter_id    UInt64,
    val             Float64
)
ENGINE = MergeTree
ORDER BY (device_id, parameter_id);

We populate this table using a Refreshable Materialized View (RMV) as a data generator:

DROP VIEW IF EXISTS lasttwo_gen;

CREATE MATERIALIZED VIEW lasttwo_gen
REFRESH EVERY 10 SECOND APPEND TO lasttwo_raw AS
SELECT
    addMilliseconds(now64(), randUniform(0, 9999)) AS ts,
    randUniform(1, 1000, 1)::UInt64 AS device_id,
    randUniform(1, 1000, 2)::UInt64 AS parameter_id,
    randNormal(100.0, 15.0) AS val
FROM numbers(1000000);

This RMV inserts a batch of 1 million rows into the IoT detail data table every 10 seconds. Because now64() returns the timestamp of RMV invocation (that is, the same timestamp for all rows), we add some random “jitter” to each timestamp. The two calls to randUniform() are using the dummy parameter technique I described in my blog about random data to prevent the optimizer from returning the same value for both.

Naïve query using window functions

If you come from a Cloud Data Warehouse background, your first approach would likely be to use window functions. To retrieve the last two values from each timeseries, you could write:

SELECT
    device_id,
    parameter_id,
    row_number() OVER (PARTITION BY device_id, parameter_id ORDER BY ts DESC) AS pos,
    val
FROM lasttwo_raw
QUALIFY pos <= 2;

using the QUALIFY clause I discussed here. This, of course, returns each value in its own row, and we have to use an outer query to pivot these values and inject each into its own column like so:

WITH cte AS (
    SELECT
        device_id,
        parameter_id,
        row_number() OVER (PARTITION BY device_id, parameter_id ORDER BY ts DESC) AS pos,
        val
    FROM lasttwo_raw
    QUALIFY pos <= 2
)
SELECT
    device_id,
    parameter_id,
    maxIf(val, pos = 1) AS last_val,    
    maxIf(val, pos = 2) AS prev_val 
FROM cte   
GROUP BY device_id, parameter_id
ORDER BY device_id, parameter_id;

This query works; I ran it after generating 251 million rows and it took roughly 37 seconds on my ClickHouse Cloud service with 3x 120 GB instances.

Interestingly, ClickHouse Cloud Assistant suggested to rewrite the query as

SELECT
    device_id,
    parameter_id,
    nth_value(val, 1) OVER w AS last_val,
    nth_value(val, 2) OVER w AS prev_val
FROM
    lasttwo_raw
WINDOW w AS (
    PARTITION BY device_id, parameter_id
    ORDER BY ts DESC
    ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
)
QUALIFY row_number() OVER w = 1
ORDER BY
    device_id,
    parameter_id;

This “optimized” query took 44 seconds. Let’s see if we can find a more idiomatic way to formulate the query in ClickHouse.

ClickHouse native aggregation

What if we could avoid calling window functions in the first place? Clickhouse has the groupArraySorted aggregator: instead of calculating a single number, it collects values in each aggregation bucket into a sorted array, keeping only the first N values.

Because we sort by ts but eventually want the val column, we need to collect both fields into a tuple; moreover, groupArraySorted sorts in ascending order, so we need to flip the sign on ts to get the latest values in the first position.

With that, our query becomes essentially a simple GROUP BY in a CTE; and some trivial logic to extract the values back:

WITH cte AS (
    SELECT
        device_id,
        parameter_id,
        groupArraySorted(2)(
            tuple(negate(toUnixTimestamp64Milli(ts)), val)
        ) AS lv
    FROM lasttwo_raw
    GROUP BY device_id, parameter_id
    ORDER BY device_id, parameter_id
)
SELECT
    device_id,
    parameter_id,
    arrayElementOrNull(lv, 1).2 AS last_val,
    arrayElementOrNull(lv, 2).2 AS prev_val
FROM cte;

With the same data set and service configuration as above, this query runs in only 2.5 seconds!

Doing things the ClickHouse way: Setting up the aggregation

But we can do even better. Going back to the beginning, let’s set up an aggregate table and (incremental) materialized view. The intermediate aggregation state is stored in an AggregateFunction object - note how the array size parameter is part of the aggregator specification and has to be included in the specification.

CREATE TABLE lasttwo_agg (
    device_id       UInt64,
    parameter_id    UInt64,
    last_values     AggregateFunction(groupArraySorted(2), Tuple(Int64, Float64))
)
ENGINE = AggregatingMergeTree
ORDER BY (device_id, parameter_id);

CREATE MATERIALIZED VIEW lasttwo_agg_mv TO lasttwo_agg AS
SELECT
    device_id,
    parameter_id,
    groupArraySortedState(2)(
        tuple(negate(toUnixTimestamp64Milli(ts)), val)
    ) AS last_values
FROM lasttwo_raw
GROUP BY device_id, parameter_id
ORDER BY device_id, parameter_id;

The aggregation is basically the same as in the previous query except for the -State modifier that we have to use because the target is an intermediate aggregate. We have to wipe out the detail table and restart the RMV data generator because the IMV picks up only newly added rows.

Then the query is reformulated to go against the aggregate, using the same aggregator with -Merge.

WITH cte AS (
    SELECT
        device_id,
        parameter_id,
        groupArraySortedMerge(2)(last_values) AS lv
    FROM lasttwo_agg
    GROUP BY device_id, parameter_id
    ORDER BY device_id, parameter_id
)
SELECT
    device_id,
    parameter_id,
    arrayElementOrNull(lv, 1).2 AS last_val,
    arrayElementOrNull(lv, 2).2 AS prev_val
FROM cte;

This query, with the same underlying data set and service configuration, runs in only 0.35 seconds!

Conclusion

We started with a question that seemed to call for a window function approach, and implemented this approach. While the result was decent, two modifications brought the query to ClickHouse speed standards:

Using ClickHouse native aggregation gave a roughly 15x speedup.
Another large boost comes from incrementally pre-aggregating data using a materialized view. The final query thus becomes more than 100x faster than the original one.

If a query in ClickHouse seems to be not quite fast, think about how to rewrite it using native functions!

“This image is taken from Page 377 of Praktisches Kochbuch für die gewöhnliche und feinere Küche” by Medical Heritage Library, Inc. is licensed under CC BY-NC-SA 2.0 .