ClickHouse Materialized Views & ReplacingMergeTree: Sub-Second Real-Time Analytics

ClickHouse excels at real-time analytical workloads. Achieving sub-second query latencies on high-cardinality, high-volume data streams often necessitates pre-aggregation and efficient deduplication. This guide details the architecture and implementation of ClickHouse Materialized Views with AggregatingMergeTree and ReplacingMergeTree engines to build a robust, real-time analytics pipeline.

Architectural Overview: Stream Aggregation & Deduplication

The core problem addressed is the need for real-time, aggregated metrics from an event stream, where events might arrive out-of-order or be duplicated. Our solution involves:

1. Raw Event Table: An append-only table storing all incoming events. This serves as the source of truth.
2. Deduplication Layer: A ReplacingMergeTree table to ensure event idempotency, handling late-arriving or replayed events.
3. Materialized View for Aggregation: An AggregatingMergeTree table, populated by a Materialized View, to pre-compute aggregates. This table stores stateful aggregates, significantly reducing query time.

This layered approach provides both data integrity and query performance.

Raw Event Ingestion with ReplacingMergeTree

First, define a raw events table. While a simple MergeTree could suffice, ReplacingMergeTree is crucial for idempotent ingestion, especially when dealing with event replay or at-least-once delivery semantics from message queues like Kafka.

CREATE TABLE IF NOT EXISTS events_raw (
    event_id UUID,
    user_id String,
    event_type LowCardinality(String),
    event_time DateTime64(3),
    value Float64,
    _version UInt64 DEFAULT 1 -- Version column for ReplacingMergeTree
) ENGINE = ReplacingMergeTree(_version)
ORDER BY (event_id, event_time)
PRIMARY KEY (event_id);

• ReplacingMergeTree(_version): This engine ensures that for any given ORDER BY key (here, event_id, event_time), only the row with the maximum _version is kept during merges. If _version is omitted, the last row by insertion order is kept.
• ORDER BY (event_id, event_time): Defines the sort key. ReplacingMergeTree uses this to identify "duplicate" rows.
• PRIMARY KEY (event_id): Optimizes point lookups and range scans on event_id.

Let's insert some sample data, including a duplicate event_id with a higher version.

INSERT INTO events_raw (event_id, user_id, event_type, event_time, value, _version) VALUES
('a0000000-0000-0000-0000-000000000001', 'user1', 'page_view', '2023-10-26 10:00:00.000', 1.0, 1),
('a0000000-0000-0000-0000-000000000002', 'user1', 'click', '2023-10-26 10:00:05.000', 0.5, 1),
('a0000000-0000-0000-0000-000000000003', 'user2', 'page_view', '2023-10-26 10:00:10.000', 1.0, 1),
('a0000000-0000-0000-0000-000000000001', 'user1', 'page_view', '2023-10-26 10:00:00.000', 1.2, 2); -- Duplicate event_id, higher version

To observe the deduplication, we need to force a merge or wait for background merges.

OPTIMIZE TABLE events_raw FINAL;

SELECT event_id, user_id, value, _version FROM events_raw ORDER BY event_id;

Output:

┌─event_id─────────────────────────────┬─user_id─┬─value─┬─_version─┐
│ a0000000-0000-0000-0000-000000000001 │ user1   │   1.2 │        2 │
│ a0000000-0000-0000-0000-000000000002 │ user1   │   0.5 │        1 │
│ a0000000-0000-0000-0000-000000000003 │ user2   │   1.0 │        1 │
└──────────────────────────────────────┴─────────┴───────┴──────────┘

Notice event_id a0000000-0000-0000-0000-000000000001 now has value 1.2 and _version 2, demonstrating the deduplication.

Materialized Views for Real-Time Aggregation

Materialized Views in ClickHouse are not pre-computed snapshots like in traditional RDBMS. Instead, they are triggers that execute a SELECT query on new data inserted into the source table and write the results into a target table. This makes them ideal for continuous, incremental aggregation.

AggregatingMergeTree for Stateful Aggregates

The target table for our Materialized View will use the AggregatingMergeTree engine. This engine stores states of aggregate functions, not their final values. When data parts are merged, these states are combined using the *Merge functions.

Define the aggregated table:

CREATE TABLE IF NOT EXISTS events_agg_mv (
    event_date Date,
    event_type LowCardinality(String),
    user_id String,
    total_value AggregateFunction(sum, Float64),
    unique_users AggregateFunction(uniqHLL12, String),
    event_count AggregateFunction(count)
) ENGINE = AggregatingMergeTree()
ORDER BY (event_date, event_type, user_id);

• AggregateFunction(sum, Float64): This special data type stores the intermediate state of the sum aggregate function.
• uniqHLL12: A highly efficient approximate distinct count algorithm, suitable for high-cardinality data.
• ORDER BY (event_date, event_type, user_id): Defines the aggregation key. All rows with the same key will be merged, and their aggregate states combined.

Creating the Materialized View

Now, create the Materialized View that populates events_agg_mv from events_raw.

CREATE MATERIALIZED VIEW IF NOT EXISTS mv_events_agg
TO events_agg_mv
AS SELECT
    toDate(event_time) AS event_date,
    event_type,
    user_id,
    sumState(value) AS total_value,
    uniqHLL12State(user_id) AS unique_users,
    countState() AS event_count
FROM events_raw
GROUP BY event_date, event_type, user_id;

• TO events_agg_mv: Specifies the target table.
• sumState(value), uniqHLL12State(user_id), countState(): These are the "state" versions of aggregate functions. They return the intermediate state of the aggregation, which AggregatingMergeTree stores.

Let's insert more data into events_raw and observe the Materialized View's effect.

INSERT INTO events_raw (event_id, user_id, event_type, event_time, value, _version) VALUES
('a0000000-0000-0000-0000-000000000004', 'user1', 'page_view', '2023-10-26 10:00:15.000', 1.0, 1),
('a0000000-0000-0000-0000-000000000005', 'user2', 'click', '2023-10-26 10:00:20.000', 0.8, 1),
('a0000000-0000-0000-0000-000000000006', 'user1', 'page_view', '2023-10-27 11:00:00.000', 1.0, 1);

Query the aggregated table. Note the use of *Merge functions to finalize the aggregates.

SELECT
    event_date,
    event_type,
    user_id,
    sumMerge(total_value) AS final_total_value,
    uniqHLL12Merge(unique_users) AS final_unique_users,
    countMerge(event_count) AS final_event_count
FROM events_agg_mv
GROUP BY event_date, event_type, user_id
ORDER BY event_date, event_type, user_id;

Output:

┌─event_date─┬─event_type─┬─user_id─┬─final_total_value─┬─final_unique_users─┬─final_event_count─┐
│ 2023-10-26 │ click      │ user1   │               0.5 │                  1 │                 1 │
│ 2023-10-26 │ click      │ user2   │               0.8 │                  1 │                 1 │
│ 2023-10-26 │ page_view  │ user1   │               2.2 │                  1 │                 2 │
│ 2023-10-26 │ page_view  │ user2   │               1.0 │                  1 │                 1 │
│ 2023-10-27 │ page_view  │ user1   │               1.0 │                  1 │                 1 │
└────────────┴────────────┴─────────┴───────────────────┴────────────────────┴───────────────────┘

The aggregates are correctly updated in real-time. The ReplacingMergeTree on events_raw ensures that if an event is re-sent with a higher version, the Materialized View will process the updated event, and the AggregatingMergeTree will correctly reflect the change upon subsequent merges.

Comparison: Normal View vs. Materialized View

Optimizing Sub-50ms Analytical Queries

Achieving sub-50ms latency requires careful consideration of table design, query patterns, and ClickHouse configuration.

1. AggregatingMergeTree ORDER BY Key: The ORDER BY clause in events_agg_mv is critical. It should match the common GROUP BY and WHERE clauses of your analytical queries. For example, if you frequently query by event_date and event_type, these should be leading columns in ORDER BY.
2. PRIMARY KEY: For AggregatingMergeTree, the PRIMARY KEY is typically a prefix of the ORDER BY key. It helps prune data parts quickly.
3. LowCardinality Data Type: Use LowCardinality(String) for columns with a limited number of distinct values (e.g., event_type). This significantly reduces storage and improves query performance due to dictionary encoding.
4. DateTime64 vs DateTime: DateTime64 offers millisecond precision, which is often necessary for event streams. Ensure your toDate() or toStartOfHour() functions align with your desired aggregation granularity.
5. FINAL Keyword: When querying AggregatingMergeTree or ReplacingMergeTree, SELECT ... FROM table FINAL ensures all merges are completed and you get the fully aggregated/deduplicated result. However, FINAL can be slow as it forces merges. For real-time dashboards, you might tolerate slightly stale data and omit FINAL, relying on background merges. For critical reports, FINAL is necessary.
6. index_granularity: This setting (default 8192) determines how many rows are in a data block for indexing. Adjusting it can impact performance, but the default is often suitable.
7. Hardware: Sufficient RAM, fast NVMe SSDs, and CPU cores are paramount for ClickHouse performance.
8. Distributed Tables: For very large datasets, use Distributed tables on top of AggregatingMergeTree to scale horizontally.

Production Gotchas & Troubleshooting

1. Materialized View Lag:

   • Symptom: events_agg_mv is not updating quickly, or queries on it show old data.
   • Cause: High ingestion rate into events_raw combined with complex MV logic or resource constraints. Materialized Views process data synchronously on insertion. If the MV query is slow, it can block insertions.
   • Fix:
     • Simplify the MV SELECT query.
     • Ensure events_raw has appropriate ORDER BY and PRIMARY KEY for efficient MV processing.
     • Scale ClickHouse resources (CPU, RAM, IO).
     • Consider using an asynchronous Materialized View (by omitting TO target_table and letting the MV create its own . table, then creating a separate AggregatingMergeTree table and inserting into it from the . table via a separate process or another MV). This decouples ingestion from aggregation but adds complexity. For most cases, synchronous MVs are preferred for simplicity and real-time guarantees.

2. ReplacingMergeTree Not Deduplicating:

   • Symptom: Duplicate event_ids persist even after OPTIMIZE TABLE FINAL.
   • Cause: Incorrect ORDER BY key. ReplacingMergeTree deduplicates based on the ORDER BY key. If event_id is not part of the ORDER BY key, or if the _version column is not correctly used, deduplication won't happen as expected.
   • Fix: Verify ORDER BY includes the unique identifier (e.g., event_id) and the version column (e.g., _version) is correctly populated and specified in the engine definition.

3. AggregatingMergeTree Query Performance:

   • Symptom: Queries on events_agg_mv are slow, even with *Merge functions.
   • Cause:
     • GROUP BY clause in the query does not align with the ORDER BY key of events_agg_mv. This forces ClickHouse to read more data than necessary.
     • Too many distinct values in the GROUP BY columns, leading to a large number of small data parts or high memory usage during merge.
     • Using FINAL unnecessarily.
   • Fix:
     • Refactor events_agg_mv ORDER BY to match common query patterns.
     • Ensure PRIMARY KEY is a prefix of ORDER BY.
     • Avoid FINAL unless strictly necessary for correctness.
     • Consider pre-aggregating further if intermediate aggregates are still too granular.

4. Disk Space Consumption:

   • Symptom: AggregatingMergeTree tables consume excessive disk space.
   • Cause: AggregateFunction states can be larger than final values, especially for functions like uniqHLL12State or groupArrayState. Also, if ORDER BY key has high cardinality, it can lead to many small data parts.
   • Fix:
     • Review aggregate functions. uniqHLL12 is space-efficient for distinct counts. groupArrayState can be very large.
     • Ensure ORDER BY key is chosen to balance aggregation granularity and data part size.
     • Implement a TTL (Time-To-Live) policy on events_raw and events_agg_mv to automatically delete old data.

-- Example TTL for events_raw (delete raw data after 30 days)
ALTER TABLE events_raw MODIFY TTL event_time + INTERVAL 30 DAY;

-- Example TTL for events_agg_mv (delete aggregates after 365 days)
ALTER TABLE events_agg_mv MODIFY TTL event_date + INTERVAL 365 DAY;

Frequently Asked Questions

1. Can I modify a Materialized View after creation? No, Materialized Views cannot be directly modified. You must DROP and CREATE them again. This is why it's crucial to design your MV carefully. If the target table schema changes, you must also drop and recreate the MV.

2. What happens if data is deleted from the source table of a Materialized View? Materialized Views only react to INSERT operations. DELETE or UPDATE operations on the source table will not automatically propagate to the Materialized View's target table. If you need to handle deletions, you'd typically implement a "soft delete" mechanism (e.g., an is_deleted flag) and filter it in your queries, or use a more complex CollapsingMergeTree or VersionedCollapsingMergeTree setup.

3. How does ReplacingMergeTree handle concurrent inserts with the same ORDER BY key? ReplacingMergeTree handles concurrent inserts by applying the replacement logic during merges. If two inserts with the same ORDER BY key but different _version values arrive concurrently, they will initially exist in separate data parts. When these parts are merged, the row with the highest _version will be retained. This ensures eventual consistency.

4. When should I use AggregatingMergeTree versus a regular MergeTree with GROUP BY? Use AggregatingMergeTree when you need to continuously aggregate data in real-time and query those aggregates frequently. It pre-computes and stores aggregate states, making queries significantly faster. Use a regular MergeTree with GROUP BY for ad-hoc aggregations on raw data where real-time performance isn't critical, or when the aggregation keys are highly dynamic and cannot be pre-defined. AggregatingMergeTree is for fixed, known aggregation patterns.

5. Are Materialized Views suitable for all types of aggregations? Materialized Views are best suited for aggregations that are additive or can be expressed using AggregateFunction states. This includes sum, count, min, max, uniqHLL12, avg (using sumState and countState), etc. Aggregations requiring access to all raw rows (e.g., quantile, median without specific AggregateFunction support) or complex window functions are generally not suitable for direct Materialized View pre-aggregation and are better run on the raw data or a more granular aggregate.