ClickHouse Streaming Patterns

Overview

Stream processing continuously ingests data from message queues into ClickHouse. Core principle: Balance throughput with latency, handle failures gracefully, ensure exactly-once delivery.

Key challenge: ClickHouse is optimized for batch inserts. Streaming requires careful tuning to avoid "too many parts" while maintaining low latency.

When to Use

Symptoms:

• Need to ingest Kafka/RabbitMQ messages into ClickHouse
• Real-time analytics dashboard (< 1 minute latency)
• IoT sensor data, application logs, clickstream
• Message rate varies (need backpressure handling)

When NOT to use:

• Database replication (PostgreSQL/MySQL) → See

  clickhouse-cdc

• Scheduled batch jobs (hourly/daily) → See

  clickhouse-patterns

• One-time data import → Use batch INSERT

Prerequisites

• Understanding of message queue semantics (Kafka offsets, RabbitMQ acks)
• Basic ClickHouse knowledge → See

  clickhouse-patterns

• Familiarity with stream processing concepts

Quick Reference

Method Selection

digraph streaming_methods {
    rankdir=TD;
    node [shape=box, style=rounded];

    start [label="Choose Method", shape=ellipse];
    native [label="ClickHouse native?", shape=diamond];
    control [label="Need custom logic?", shape=diamond];

    kafka_engine [label="Kafka Engine\n(native, simple)"];
    kafka_connect [label="Kafka Connect\n(custom transform)"];
    custom_service [label="Custom Service\n(full control)"];

    start -> native;
    native -> kafka_engine [label="Kafka"];
    native -> control [label="RabbitMQ/Kinesis"];
    control -> kafka_connect [label="no"];
    control -> custom_service [label="yes"];
}

Method	Best For	Pros	Cons
Kafka Engine	Simple Kafka → CH	Native, no code	Limited transformation
Custom Service	Complex logic, any queue	Full control	Must handle failures

Critical Settings

Setting	Low Latency	High Throughput
Batch size	1000 rows	50000 rows
Flush interval	500ms	5000ms
Target latency	< 1 second	5-10 seconds

Pattern 1: ClickHouse Kafka Engine

Setup

-- 1. Kafka staging table
CREATE TABLE events_queue (
    timestamp UInt64,
    user_id String,
    event_type String,
    properties String
) ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'kafka:9092',
    kafka_topic_list = 'events',
    kafka_group_name = 'clickhouse_consumers',
    kafka_format = 'JSONEachRow',
    kafka_num_consumers = 4,           -- Match Kafka partitions
    kafka_max_block_size = 65536,      -- Batch size
    kafka_poll_timeout_ms = 1000;      -- Wait time

-- 2. Target table
CREATE TABLE events (
    date Date,
    timestamp DateTime,
    user_id String,
    event_type LowCardinality(String),
    properties String
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(date)
ORDER BY (date, user_id, timestamp);

-- 3. Materialized view
CREATE MATERIALIZED VIEW events_mv TO events AS
SELECT
    toDate(toDateTime(timestamp)) AS date,
    toDateTime(timestamp) AS timestamp,
    user_id, event_type, properties
FROM events_queue;

Tuning Parameters

SETTINGS
    kafka_max_block_size = 65536,        -- Batch size (larger = higher latency)
    kafka_poll_timeout_ms = 1000,        -- Poll wait (longer = better batching)
    kafka_num_consumers = 4,             -- Parallel consumers
    kafka_skip_broken_messages = 100,    -- Skip errors (use carefully)
    kafka_commit_every_batch = 1;        -- Commit frequency

Monitoring

-- Consumer status
SELECT database, table, consumer_id, exceptions, last_exception_time
FROM system.kafka_consumers;

-- Consumption lag
SELECT table, partition_id,
       current_offset - last_committed_offset AS lag
FROM system.kafka_consumers;

Pattern 2: Custom Service (Node.js)

Use Case

• Complex transformations
• Multiple sources (RabbitMQ, Kinesis)
• Custom error handling

Implementation

import { Kafka } from 'kafkajs';
import { ClickHouse } from 'clickhouse';

class StreamProcessor {
  private buffer: any[] = [];
  private readonly BATCH_SIZE = 5000;
  private readonly FLUSH_INTERVAL_MS = 1000;

  async start() {
    const consumer = kafka.consumer({ groupId: 'events-processor' });
    await consumer.subscribe({ topic: 'events' });

    setInterval(() => this.flush(), this.FLUSH_INTERVAL_MS);

    await consumer.run({
      eachBatchAutoResolve: false,
      eachBatch: async ({ batch, resolveOffset, heartbeat }) => {
        for (const message of batch.messages) {
          const event = JSON.parse(message.value.toString());

          this.buffer.push({
            date: new Date(event.timestamp).toISOString().split('T')[0],
            timestamp: new Date(event.timestamp),
            user_id: event.userId,
            event_type: event.type,
            properties: JSON.stringify(event.properties)
          });

          if (this.buffer.length >= this.BATCH_SIZE) await this.flush();

          await resolveOffset(message.offset);
          await heartbeat();
        }
      }
    });
  }

  async flush() {
    if (this.buffer.length === 0) return;

    try {
      await clickhouse.insert('events', this.buffer);
      console.log(`Flushed ${this.buffer.length} events`);
      this.buffer = [];
    } catch (err) {
      await this.retryWithBackoff(this.buffer);
    }
  }

  async retryWithBackoff(batch: any[], attempt = 1) {
    if (attempt > 3) {
      console.error('Max retries, sending to DLQ');
      return;
    }

    const delay = Math.min(1000 * Math.pow(2, attempt), 10000);
    await new Promise(r => setTimeout(r, delay));

    try {
      await clickhouse.insert('events', batch);
    } catch (err) {
      await this.retryWithBackoff(batch, attempt + 1);
    }
  }
}

Exactly-Once Semantics

Challenge

Kafka at-least-once + ClickHouse idempotency = exactly-once

Solution 1: Deduplication Table

CREATE TABLE processed_messages (
    message_id String,
    processed_at DateTime
) ENGINE = MergeTree()
ORDER BY message_id
TTL processed_at + INTERVAL 7 DAY;

-- Insert only new messages
INSERT INTO events
SELECT * FROM events_staging
WHERE message_id NOT IN (SELECT message_id FROM processed_messages);

Solution 2: ReplacingMergeTree

CREATE TABLE events (
    message_id String,  -- Kafka offset or UUID
    timestamp DateTime,
    user_id String
) ENGINE = ReplacingMergeTree()
ORDER BY message_id;

SELECT * FROM events FINAL WHERE user_id = 'user-123';

Backpressure Handling

Bounded Buffer Pattern

class BackpressureBuffer {
  private buffer: any[] = [];
  private readonly MAX_BUFFER_SIZE = 100000;
  private processing = false;

  async add(item: any) {
    while (this.buffer.length >= this.MAX_BUFFER_SIZE) {
      await new Promise(r => setTimeout(r, 100));
    }
    this.buffer.push(item);
  }

  async tryFlush() {
    if (this.processing || this.buffer.length < BATCH_SIZE) return;

    this.processing = true;
    const batch = this.buffer.splice(0, BATCH_SIZE);

    try {
      await clickhouse.insert('events', batch);
    } finally {
      this.processing = false;
    }
  }
}

Pause Consumer Pattern

consumer.run({
  eachBatch: async ({ batch, pause }) => {
    if (buffer.length > MAX_BUFFER_SIZE) {
      const pauseHandle = pause();
      await flush();
      pauseHandle.resume();
    }
  }
});

Performance Optimization

Async Inserts

SET async_insert = 1;
SET wait_for_async_insert = 0;
SET async_insert_max_data_size = 10485760;  -- 10MB
SET async_insert_busy_timeout_ms = 1000;    -- 1s

Parallel Consumers

-- Match number to Kafka partitions
SETTINGS kafka_num_consumers = 8;

Common Mistakes

Mistake	Why It Fails	Fix
Small batches	Too many parts	Minimum 1000 rows
No backpressure	Memory overflow	Bounded buffer + pause
Sync inserts	Low throughput	Use async_insert = 1
No error handling	Lost messages	Retry + DLQ
Single consumer	Low throughput	Parallel = partitions

Monitoring

-- Throughput
SELECT toStartOfMinute(now()) AS minute, count() AS events_per_minute
FROM events
WHERE timestamp >= now() - INTERVAL 5 MINUTE
GROUP BY minute;

-- Lag analysis
SELECT
    quantile(0.50)(now() - timestamp) AS median_lag,
    quantile(0.95)(now() - timestamp) AS p95_lag
FROM events
WHERE timestamp >= now() - INTERVAL 1 MINUTE;

-- Errors
SELECT toStartOfMinute(event_time) AS minute, count() AS errors
FROM system.kafka_consumers
WHERE last_exception_time >= now() - INTERVAL 1 HOUR
GROUP BY minute;

Best Practices

Design:

• Use Kafka Engine for simple pipelines
• Custom service for complex transformations
• Always implement deduplication

Performance:

• Batch size: 1000-50000 (tune for latency)
• Flush interval: 500ms-5s
• Parallel consumers = Kafka partitions

Reliability:

• Bounded buffer for backpressure
• Exponential backoff retry
• Dead letter queue for failures
• Monitor lag continuously

Red Flags

• ❌ "Insert per message" → Too many parts
• ❌ "Unbounded buffer" → Memory overflow
• ❌ "No lag monitoring" → Stale data
• ❌ "Ignore errors" → Data loss
• ❌ "Sync inserts" → Low throughput

When to Escalate

• Lag > 1 minute despite tuning
• Memory exhaustion with backpressure
• "Too many parts" errors
• Multi-datacenter streaming needed

Resources: ClickHouse Kafka engine docs, system.kafka_consumers table

Related Skills

• ClickHouse fundamentals: See

  clickhouse-patterns

• Database replication: See

  clickhouse-cdc

---

Remember: Balance latency, throughput, and reliability. Monitor lag, implement backpressure, tune batch sizes for your SLA.