ClickHouse data engineers

ClickHouse®'s storage model is fundamentally different from the databases most data engineers build pipelines into. It's not an RDBMS. It's not a message queue. Updates and deletes are expensive and asynchronous. Row-by-row inserts will break it. And the schema decisions you make at design time, partition key, sort key, table engine, are difficult to change later without rebuilding the table.

Get those decisions right and ClickHouse rewards you with ingestion throughput and query speed that's difficult to match anywhere else. Get them wrong and you'll spend weeks fighting the merge scheduler or rebuilding tables in production.

This post covers the decisions that matter most for data engineers: table engine selection, ingestion batching, CDC patterns, streaming ETL, ELT architecture with materialized views, schema evolution, and monitoring pipeline health.

Table engine selection

All production ClickHouse tables use a MergeTree variant. The right variant depends on what the data represents and how it changes:

Use MergeTree as the default. If your pipeline is append-only (events, logs, metrics), standard MergeTree is the right choice. It's simple, predictable, and fast. Don't reach for a more complex engine until you have a concrete need.

Use ReplacingMergeTree for mutable records. CDC streams from Postgres or MySQL produce inserts and updates for the same primary key. ReplacingMergeTree deduplicates rows with the same sort key during background merges, keeping the row with the highest version:

CREATE TABLE users
(
    id          UInt64,
    email       String,
    plan        LowCardinality(String),
    updated_at  DateTime,
    _deleted    UInt8 DEFAULT 0
)
ENGINE = ReplacingMergeTree(updated_at)
ORDER BY id;

Background merges are asynchronous, so deduplication isn't immediate. For exact results, query with FINAL (slower) or use the argMax pattern (faster):

-- argMax pattern: exact latest state without FINAL
SELECT
    id,
    argMax(email, updated_at)   AS email,
    argMax(plan, updated_at)    AS plan
FROM users
WHERE _deleted = 0
GROUP BY id;

Use AggregatingMergeTree for pre-computed metrics. When you're building rollup tables that materialized views write to, AggregatingMergeTree stores partial aggregate states (countState, sumState, uniqState) and merges them efficiently. Queries use the corresponding -Merge functions:

SELECT
    day,
    countMerge(events)      AS total_events,
    uniqMerge(users)        AS unique_users,
    sumMerge(revenue)       AS total_revenue
FROM events_daily
WHERE day >= today() - 30
GROUP BY day;

Ingestion: the batching requirement

ClickHouse writes data in "parts", one part per insert. Background merges consolidate parts over time. If you insert row by row, you create one part per row, which quickly overwhelms the merge scheduler and causes "too many parts" errors that block further inserts.

Always batch inserts. The practical targets are 1,000–100,000 rows per insert or 1–10MB per batch. Anything below 1,000 rows should be buffered before sending.

For streaming pipelines, flush on time interval OR row count, whichever comes first:

class ClickHouseBatcher:
    def __init__(self, flush_interval=5.0, min_rows=1000):
        self.buffer = []
        self.last_flush = time.time()
        self.flush_interval = flush_interval
        self.min_rows = min_rows

    def add(self, row):
        self.buffer.append(row)
        if (len(self.buffer) >= self.min_rows or
                time.time() - self.last_flush >= self.flush_interval):
            self.flush()

    def flush(self):
        if not self.buffer:
            return
        batch, self.buffer = self.buffer[:], []
        client.insert("events", batch, column_names=[...])
        self.last_flush = time.time()

For async insert support (ClickHouse buffers server-side), enable async_insert=1 in your connection settings. ClickHouse collects inserts in memory and flushes them to disk in batches automatically, allowing your application to send single rows without part fragmentation.

Monitor part counts as part of your pipeline health checks:

SELECT
    table,
    partition,
    count() AS active_parts
FROM system.parts
WHERE database = 'default'
  AND active = 1
GROUP BY table, partition
HAVING active_parts > 100
ORDER BY active_parts DESC;

If any partition exceeds 300 active parts, your pipeline is inserting too frequently or in batches that are too small. This is the single most common cause of ClickHouse performance degradation in production.

Streaming ETL from Kafka

The standard Kafka-to-ClickHouse pattern uses the Kafka table engine as a staging layer, with a materialized view handling the transformation and a MergeTree table as the final destination. See how to stream data from Kafka to ClickHouse for the full setup.

The materialized view between the Kafka staging table and the MergeTree destination is where lightweight ETL happens:

CREATE MATERIALIZED VIEW etl_raw_to_events TO events AS
SELECT
    toDateTime(JSONExtractString(raw_data, 'ts'))        AS event_time,
    JSONExtractString(raw_data, 'type')                  AS event_type,
    toUInt32OrZero(JSONExtractString(raw_data, 'uid'))   AS user_id,
    upper(JSONExtractString(raw_data, 'country'))        AS country,
    toDecimal64OrZero(
        JSONExtractString(raw_data, 'revenue'), 4)       AS revenue
FROM events_kafka;

Keep transformation logic in the materialized view deterministic and lightweight: type casting, field extraction, normalization. Push heavy transforms (ML inference, DOM parsing, external lookups) to asynchronous workers outside the ingestion path.

For malformed records, add a dead letter queue to capture parsing failures without dropping them silently:

-- Kafka table with error mode enabled
CREATE TABLE events_kafka (...) ENGINE = Kafka
SETTINGS kafka_handle_error_mode = 'stream', ...;

-- Dead letter queue captures malformed rows
CREATE MATERIALIZED VIEW events_dlq TO error_events AS
SELECT now() AS ingested_at, _raw_message, _error
FROM events_kafka
WHERE _error != '';

CDC pipelines

Change data capture from Postgres or MySQL into ClickHouse is a common data engineering pattern. The CDC stream produces INSERT, UPDATE, and DELETE events for every committed transaction. ClickHouse needs to translate these into its append-first storage model.

ReplacingMergeTree with a version column is the standard approach. Every CDC event becomes an INSERT with an incrementing _version:

CREATE TABLE orders
(
    id          UInt64,
    status      LowCardinality(String),
    amount      Decimal(18, 2),
    updated_at  DateTime64(3),
    _version    UInt64,
    _deleted    UInt8 DEFAULT 0
)
ENGINE = ReplacingMergeTree(_version)
ORDER BY id;

For DELETE events, your CDC pipeline inserts a row with _deleted = 1 and a higher _version. Every query filters WHERE _deleted = 0 using the argMax pattern.

Batch CDC events before inserting. A flush interval of 1–5 seconds with a minimum batch of 1,000 rows is a good starting point. Inserting one CDC event at a time creates one part per change and will cause merge backlog at any meaningful change rate.

ELT architecture with materialized views

The ClickHouse equivalent of a dbt transformation layer is materialized views. Unlike dbt, which runs scheduled batch jobs, ClickHouse materialized views run incrementally at insert time: transformations happen as data arrives, not on a cron schedule.

The architecture is:

Raw ingestion tables (bronze)
        |
Materialized views (transformation)
        |
Rollup / cleaned tables (silver/gold)
        |
Endpoint Pipes (serving)

Land raw data first. Then define materialized views that extract structured fields, apply business logic, and write to clean tables:

-- Silver layer: clean events from raw JSON
CREATE MATERIALIZED VIEW events_clean_mv TO events_clean AS
SELECT
    toDateTime(JSONExtractString(raw, 'ts'))             AS timestamp,
    JSONExtractString(raw, 'event')                      AS event_type,
    JSONExtractString(raw, 'org_id')                     AS org_id,
    JSONExtractUInt(raw, 'user_id')                      AS user_id,
    JSONExtractFloat(raw, 'revenue')                     AS revenue
FROM events_raw
WHERE JSONExtractString(raw, 'event') != '';

-- Gold layer: daily rollup for dashboards
CREATE MATERIALIZED VIEW events_daily_mv TO events_daily AS
SELECT
    toDate(timestamp)        AS day,
    org_id,
    event_type,
    count()                  AS events,
    uniqState(user_id)       AS unique_users,
    sumState(revenue)        AS revenue
FROM events_clean
GROUP BY day, org_id, event_type;

Each layer only processes new data as it arrives. There's no scheduled job to maintain, no job failure to monitor. The transformation is continuous and exactly reflects the state of the raw data. If you need to rebuild a layer, you can insert historical data into the bronze table and the materialized views will process it automatically.

Schema evolution

ClickHouse supports adding nullable columns without rebuilding the table:

ALTER TABLE events ADD COLUMN sdk_version Nullable(String);
ALTER TABLE events ADD COLUMN platform LowCardinality(Nullable(String)) DEFAULT 'unknown';

For schema migrations while keeping streaming ingestion running, the order of operations matters: add the column to the destination table first, then update the materialized view to populate it. Rolling this back requires dropping and recreating the materialized view.

Changing sort keys or partition keys requires rebuilding the table. Design your partition key (typically toYYYYMM(timestamp)) and sort key (typically tenant/org columns followed by time) with 3–5 years of growth in mind. These are the decisions that are expensive to undo.

Renaming columns in ClickHouse is a lightweight operation in recent versions, but renaming a sort key column still requires a full table rebuild. When in doubt, add new columns rather than rename existing ones.

Monitoring pipeline health

A healthy data engineering pipeline into ClickHouse needs to track three things: insert throughput, merge backlog, and query latency on the tables you're writing to.

Monitor recent insert rates and errors from system.query_log:

SELECT
    toStartOfMinute(event_time) AS minute,
    count()                     AS inserts,
    sum(exception != '')        AS errors,
    avg(query_duration_ms)      AS avg_duration_ms
FROM system.query_log
WHERE type = 'QueryFinish'
  AND query_kind = 'Insert'
  AND event_time >= now() - INTERVAL 1 HOUR
GROUP BY minute
ORDER BY minute DESC;

Track background merge throughput to detect backlog:

SELECT
    table,
    count()                         AS active_merges,
    sum(rows_read)                  AS total_rows_merging,
    avg(elapsed)                    AS avg_merge_seconds
FROM system.merges
GROUP BY table
ORDER BY total_rows_merging DESC;

If merge throughput consistently lags insert throughput, you need to reduce insert frequency, increase batch size, or reduce the number of materialized views writing to the same destination table.

Tinybird for data engineers

Tinybird is managed ClickHouse with the data engineering workflow built in. Define schemas as .datasource files and transformations as SQL Pipes, commit both to Git, and deploy with tb deploy. The Events API handles batching and part management automatically. Kafka connectors wire up Kafka topics to datasources without consumer code.

The Tinybird Python SDK covers ingestion patterns in detail, and the real-time data ingestion docs walk through schema design for streaming workloads.

For data engineers who want the ClickHouse engine without the cluster operations, the pipeline from raw events to queryable API endpoint can be shipped in hours rather than days.