From Kafka Events to Sub-Second Queries — Designing a Tens-of-Thousands-per-Second Aggregation Pipeline with ClickHouse Materialized Views and Projections

When the first real-time dashboard goes live, there's one question that almost always follows: "Why are queries so slow?" You pump tens of thousands of events per second from Kafka into storage, and then face the reality of a dashboard SELECT taking 10 seconds. Adding more indexes and optimizing aggregate queries doesn't fix the root cause. The bottleneck is the architecture itself — aggregating billions of rows at query time.

ClickHouse addresses this problem head-on with three tools. Kafka Engine Table lets you declare a Consumer in SQL without a separate connector; Materialized Views complete aggregation the moment an event arrives; and Projections deliver optimal layouts for diverse filter patterns without changing a single query. Wingify's migration from PostgreSQL to this architecture — which cut their infrastructure costs by 80% — starts from exactly this same principle.

This article walks through the design principles and real code of the Kafka → Kafka Engine Table → Materialized View → AggregatingMergeTree pipeline, step by step.

Core Concepts

The 3-Layer Pipeline Architecture

The heart of ClickHouse real-time aggregation is a shift in when aggregation happens. In a traditional OLAP architecture, raw data is aggregated when a query arrives. ClickHouse's pipeline finishes aggregation at insert time. Dashboard SELECTs scan only pre-aggregated results, so response time stays consistent even at massive scale.

In practice, a single Kafka Engine Table acts as the source, with two Materialized Views attached to it. One stores the raw data as an audit log; the other aggregates immediately and writes to an AggregatingMergeTree.

Here's a summary of each layer's role:

Kafka Engine Table — Declaring a Consumer in a Single SQL Statement

The Kafka Engine Table is a native integration where ClickHouse itself acts as a Kafka Consumer. You can declare the Consumer in just a few DDL lines, with no separate Kafka Connect cluster required.

CREATE TABLE events_kafka
(
    event_id    String,
    user_id     UInt64,
    event_type  LowCardinality(String),
    page_url    String,
    country     LowCardinality(String),
    ts          DateTime64(3)
)
ENGINE = Kafka
SETTINGS
    kafka_broker_list          = 'kafka1:9092,kafka2:9092',
    kafka_topic_list           = 'user-events',
    kafka_group_name           = 'ch-events-consumer',
    kafka_format               = 'JSONEachRow',
    kafka_num_consumers        = 4,
    kafka_skip_broken_messages = 100;

kafka_num_consumers and partition count: The number of Kafka topic partitions is the upper bound for kafka_num_consumers. Setting more consumers than partitions leaves the excess idle. kafka_num_consumers = 4 is reasonable for an 8-partition topic, but kafka_num_consumers = 16 wastes half of them. Changing this value requires a ClickHouse restart, so align it with your partition count during initial design.

A note on kafka_skip_broken_messages: This setting skips up to the specified number of malformed messages at the cost of data loss. Leaving it in production code without explanation can cause data to silently disappear. In production, either leave it at the default (0), or if you do set it, pair it with an alert that monitors the skipped message count in the system.kafka_consumers view.

One critical point: this table does not store data. Running a SELECT reads from the Kafka stream and consumes offsets. Running SELECT * FROM events_kafka in production will drain offsets, so data inspection should be done with a separate consumer group or only in a development environment.

Comparison of 3 Kafka Integration Methods

Materialized View — The Engine of Insert-Time Aggregation

A Materialized View is an INSERT-time trigger that fires automatically the moment data arrives in the Kafka Engine Table. If queries took 10 seconds because aggregation happened at SELECT time, Materialized Views move that work to INSERT time.

-- Persistent table for storing actual data
CREATE TABLE events_hourly_agg
(
    event_type   LowCardinality(String),
    country      LowCardinality(String),
    hour         DateTime,
    event_count  AggregateFunction(count),
    unique_users AggregateFunction(uniq, UInt64)
)
ENGINE = AggregatingMergeTree()
ORDER BY (event_type, country, hour)
TTL hour + INTERVAL 90 DAY;
 
-- Materialized View sourced from the Kafka Engine table
CREATE MATERIALIZED VIEW mv_hourly TO events_hourly_agg AS
SELECT
    event_type,
    country,
    toStartOfHour(ts)  AS hour,
    countState()       AS event_count,
    uniqState(user_id) AS unique_users
FROM events_kafka
GROUP BY event_type, country, hour;

Using the TO keyword to explicitly specify the target table is strongly recommended. Without TO, a hidden internal table named .inner.mv_hourly is created automatically, making schema changes and migrations far more painful.

AggregatingMergeTree — The State/Merge Pattern

AggregatingMergeTree stores the intermediate state of aggregate functions and merges them during background merges. Unlike SummingMergeTree, which only supports numeric summation, it can accurately merge complex functions like uniq, avg, and quantile.

-- Write: store intermediate state using *State functions (inside Materialized View)
SELECT
    countState()       AS event_count,
    uniqState(user_id) AS unique_users,
    sumState(revenue)  AS total_revenue
FROM events_kafka
GROUP BY event_type, country, hour;
 
-- Read: merge state into final results using *Merge functions
SELECT
    event_type,
    country,
    hour,
    countMerge(event_count)   AS total_events,
    uniqMerge(unique_users)   AS dau,
    sumMerge(total_revenue)   AS revenue
FROM events_hourly_agg
WHERE hour >= now() - INTERVAL 24 HOUR
GROUP BY event_type, country, hour
ORDER BY hour DESC;

*Merge functions correctly combine intermediate states scattered across multiple parts. Even without GROUP BY, countMerge(event_count) alone computes the correct grand total. The issue isn't correctness — it's semantic intent. If you want counts per event type but omit GROUP BY event_type, you get a single correct but useless grand total.

Selecting an AggregateFunction column without a *Merge function returns a binary blob. A query like SELECT event_count FROM events_hourly_agg returns unreadable values.

Alternative to *Merge — the FINAL keyword: Besides GROUP BY + *Merge, you can use SELECT ... FROM t FINAL to force-merge unmerged parts at read time. However, FINAL reads all parts and merges them synchronously, which is expensive on large tables. For dashboard queries against aggregation tables, GROUP BY + *Merge is more efficient; FINAL is better suited to small tables or one-off validation.

Projections — Layout Optimization Without Changing Queries

Projections are auxiliary data layouts stored physically alongside the base table's data. They optimize queries that frequently filter on columns other than the PRIMARY KEY. Since the query optimizer selects them automatically, existing queries require zero changes to benefit.

-- Base table ORDER BY: (event_type, country, hour)
-- Add a Projection to optimize filtering by country
ALTER TABLE events_hourly_agg
    ADD PROJECTION proj_by_country
    (
        SELECT *
        ORDER BY (country, hour, event_type)
    );
 
-- Backfill the Projection onto existing data
ALTER TABLE events_hourly_agg MATERIALIZE PROJECTION proj_by_country;
 
-- From here, the optimizer automatically selects proj_by_country for this query
SELECT hour, countMerge(event_count) AS events
FROM events_hourly_agg
WHERE country = 'KR'
  AND hour >= now() - INTERVAL 7 DAY
GROUP BY hour;

The most reliable way to confirm a Projection was actually selected is the projections column in system.query_log.

SELECT query, projections
FROM system.query_log
WHERE query LIKE '%events_hourly_agg%'
  AND type = 'QueryFinish'
ORDER BY event_time DESC
LIMIT 5;

For a quick hint during query planning, EXPLAIN indexes = 1 SELECT ... is also available. Note that this shows indexes the optimizer considered, not necessarily the ones used at runtime. For verification purposes, system.query_log is more authoritative.

Because every INSERT also writes Projection data, there is a write overhead. Apply Projections selectively to filter patterns that are genuinely frequent.

Practical Application

Scenario: Real-Time User Behavior Event Aggregation Dashboard

The marketing team wants to see the following in real time:

• Event counts by event type, country, and time slot
• Hourly DAU (Daily Active Users)
• Sub-second responses even when filtering by country

Step 1: Set Up the Kafka Engine Source Table and Raw Storage Table

-- Kafka Engine source table (streaming source that does not store data)
CREATE TABLE events_kafka
(
    event_id    String,
    user_id     UInt64,
    event_type  LowCardinality(String),
    page_url    String,
    country     LowCardinality(String),
    ts          DateTime64(3)
)
ENGINE = Kafka
SETTINGS
    kafka_broker_list   = 'kafka1:9092,kafka2:9092',
    kafka_topic_list    = 'user-events',
    kafka_group_name    = 'ch-events-consumer',
    kafka_format        = 'JSONEachRow',
    kafka_num_consumers = 4;
    -- Leave kafka_skip_broken_messages at the default (0)
    -- Only set it explicitly when data loss is acceptable and justified
 
-- Raw table for audit log storage
-- Use ReplicatedMergeTree in a cluster environment with ZooKeeper/ClickHouse Keeper configured
-- For a single node, replace with MergeTree()
CREATE TABLE events_raw
(
    event_id    String,
    user_id     UInt64,
    event_type  LowCardinality(String),
    page_url    String,
    country     LowCardinality(String),
    ts          DateTime64(3)
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events_raw', '{replica}')
ORDER BY (ts, event_type)
TTL ts + INTERVAL 30 DAY;
 
-- MV connecting events_kafka → events_raw (for raw storage)
CREATE MATERIALIZED VIEW mv_raw TO events_raw AS
SELECT *
FROM events_kafka;

Step 2: Design the Aggregation Table and Add a Projection

CREATE TABLE events_hourly_agg
(
    event_type   LowCardinality(String),
    country      LowCardinality(String),
    hour         DateTime,
    event_count  AggregateFunction(count),
    unique_users AggregateFunction(uniq, UInt64)
)
ENGINE = AggregatingMergeTree()
ORDER BY (event_type, country, hour)
TTL hour + INTERVAL 90 DAY;
 
-- Projection to optimize filtering by country
ALTER TABLE events_hourly_agg
    ADD PROJECTION proj_by_country
    (SELECT * ORDER BY (country, hour, event_type));
 
ALTER TABLE events_hourly_agg
    MATERIALIZE PROJECTION proj_by_country;

Step 3: Connect the Aggregation Pipeline with a Materialized View

-- MV connecting events_kafka → events_hourly_agg (for aggregation)
CREATE MATERIALIZED VIEW mv_hourly TO events_hourly_agg AS
SELECT
    event_type,
    country,
    toStartOfHour(ts)  AS hour,
    countState()       AS event_count,
    uniqState(user_id) AS unique_users
FROM events_kafka
GROUP BY event_type, country, hour;

At this point, two MVs (mv_raw and mv_hourly) are both sourced from events_kafka. Every time a batch arrives from Kafka, both MVs are triggered and raw storage and aggregation are processed simultaneously.

Step 4: Dashboard Queries

-- Event type statistics for the last 24 hours
SELECT
    event_type,
    countMerge(event_count)  AS total_events,
    uniqMerge(unique_users)  AS dau
FROM events_hourly_agg
WHERE hour >= now() - INTERVAL 24 HOUR
GROUP BY event_type
ORDER BY total_events DESC;
 
-- Hourly breakdown by country (proj_by_country selected automatically)
SELECT
    hour,
    countMerge(event_count) AS events
FROM events_hourly_agg
WHERE country = 'KR'
  AND hour >= now() - INTERVAL 7 DAY
GROUP BY hour
ORDER BY hour;

Multi-Level Rollups via Materialized View Chaining

Chaining Materialized Views lets you complete multi-level aggregation entirely inside ClickHouse. This covers cases where only simple GROUP BY-based time-to-day rollups are needed. Complex transformations requiring stream JOINs or window functions are not supported in Materialized View SQL — in those cases, the practical approach is to pre-process in Kafka Streams or Flink and then load into ClickHouse.

-- Daily rollup storage table
CREATE TABLE events_daily_agg
(
    event_type   LowCardinality(String),
    country      LowCardinality(String),
    day          Date,
    event_count  AggregateFunction(count),
    unique_users AggregateFunction(uniq, UInt64)
)
ENGINE = AggregatingMergeTree()
ORDER BY (event_type, country, day);
 
-- Daily rollup view sourced from the hourly aggregation results
CREATE MATERIALIZED VIEW mv_daily TO events_daily_agg AS
SELECT
    event_type,
    country,
    toDate(hour)                  AS day,
    countMergeState(event_count)  AS event_count,
    uniqMergeState(unique_users)  AS unique_users
FROM events_hourly_agg
GROUP BY event_type, country, day;

The key here is the use of countMergeState() in the rollup view. It merges the intermediate state already stored as AggregateFunction type, then saves the result as a new State — this is how chaining continues.

Trade-offs

Advantages

Wingify migrated from PostgreSQL to this architecture and cut infrastructure costs by 80% (see the reference link below).

Disadvantages and Key Considerations

Common Pitfalls in Practice

1. Running SELECT directly on the Kafka Engine table

Running SELECT * FROM events_kafka in production consumes Kafka offsets. Data inspection should be done with a separate consumer group or only in a development environment.

2. Confusing GROUP BY and *Merge functions when reading from AggregatingMergeTree

-- Grand total is computed but there's no per-key breakdown
-- countMerge() returns the correct value, but per-event-type separation is lost
SELECT countMerge(event_count) FROM events_hourly_agg;
 
-- GROUP BY is required to see counts broken down by event type
SELECT event_type, countMerge(event_count)
FROM events_hourly_agg
GROUP BY event_type;
 
-- Selecting an AggregateFunction column without *Merge returns a binary blob
SELECT event_count FROM events_hourly_agg;  -- unreadable value

countMerge(event_count) computes the correct grand total even without GROUP BY. The issue isn't correctness — it's semantic intent. If you want per-event-type counts but omit GROUP BY, you're left with a single correct but useless grand total.

3. Creating a Materialized View without TO

-- Pattern to avoid (hidden internal table is created automatically)
CREATE MATERIALIZED VIEW mv_hourly AS SELECT ...;
 
-- Recommended pattern (explicitly specify the target table)
CREATE MATERIALIZED VIEW mv_hourly TO events_hourly_agg AS SELECT ...;

Without TO, a hidden internal table named .inner.mv_hourly is created, making schema changes and migrations far more painful.

Closing Thoughts

Shifting aggregation from query time to insert time — that single architectural decision turns a sluggish dashboard into sub-second responses. The Kafka Engine Table accepts the event stream without an external consumer; Materialized Views aggregate data the instant it arrives; and AggregatingMergeTree stores and merges that intermediate state efficiently. Add Projections on top, and a single table responds optimally to multiple filter patterns without changing a single query.

3 Steps to Get Started:

1. Spin up ClickHouse with Docker and try declaring a Kafka Engine table — You can integrate with a real Kafka using Confluent Cloud or a free Redpanda plan. Start with the JSONEachRow format and leave kafka_skip_broken_messages at the default (0).

2. Learn the *State/*Merge pattern using AggregatingMergeTree + Materialized View — Start by practicing with just the countState()/countMerge() pair. Once comfortable, expand to uniqState, quantileState, and others.

3. Add one Projection and verify it's actually being used — After adding a Projection and running a query, check the projections column in system.query_log to confirm the Projection was selected. If the column is empty, check whether your WHERE condition matches the leading column in the Projection's ORDER BY.

References

• Integrating Kafka with ClickHouse — Official Docs
• Kafka Table Engine — Official Docs
• Materialized Views versus Projections — Official Docs
• AggregatingMergeTree — Official Docs
• Kafka to ClickHouse: 3 Ingestion Methods Compared — GlassFlow
• Real-Time Analytics with ClickHouse and Kafka — GlassFlow
• How Braze rebuilt its real-time analytics pipeline with ClickHouse Cloud
• From Postgres to ClickHouse: Achieving Real-Time Aggregations — Wingify Engineering
• Simplifying Real-Time Data Pipelines: How ClickHouse Replaced Flink — Medium
• How to Compare Projections vs Materialized Views in ClickHouse — OneUptime Blog
• ClickHouse Kafka Engine FAQ — Altinity
• ClickHouse Projections Complete Guide — ObsessionDB
• AggregatingMergeTree in ClickHouse: Pre-Aggregated Storage Explained — Pulse
• ClickHouse Query Performance Optimization: Complete 2026 Guide — e6data
• ClickHouse 2025 Roundup — Official Blog