ClickHouse Data Ingestion

Overview

Build data ingestion pipelines into ClickHouse from HTTP webhooks, Kafka, and

streaming sources with proper batching, deduplication, and error handling.

The core rule: ClickHouse hates one-row-at-a-time inserts — buffer events and

flush them in batches. This skill covers four ingestion paths (application-side

webhook receiver, server-side Kafka engine, managed ClickPipes, and HTTP bulk

loads) plus idempotent dedup and insert monitoring.

Prerequisites

• A ClickHouse table with an appropriate engine already exists (a MergeTree

variant, e.g. analytics.events) — see clickhouse-core-workflow-a.

• The @clickhouse/client package is installed and connected via

CLICKHOUSE_HOST.

• For the Kafka paths, a reachable Kafka broker and topic.

Instructions

Step 1: Webhook Receiver with Batched Inserts

Buffer incoming events in memory, flush on a size threshold or a timer, and

re-queue the batch on failure so no event is lost. This is the application-side

core of the skill:

import express from 'express';
import { createClient } from '@clickhouse/client';

const client = createClient({ url: process.env.CLICKHOUSE_HOST! });
const app = express();
app.use(express.json());

// Buffer for batching — ClickHouse hates one-row-at-a-time inserts
const buffer: Record<string, unknown>[] = [];
const BATCH_SIZE = 5_000;
const FLUSH_INTERVAL_MS = 5_000;

async function flushBuffer() {
  if (buffer.length === 0) return;
  const batch = buffer.splice(0, buffer.length);

  try {
    await client.insert({
      table: 'analytics.events',
      values: batch,
      format: 'JSONEachRow',
    });
    console.log(`Flushed ${batch.length} events to ClickHouse`);
  } catch (err) {
    console.error('Insert failed, re-queuing:', (err as Error).message);
    buffer.unshift(...batch);  // Put back at front for retry
  }
}

// Flush periodically
setInterval(flushBuffer, FLUSH_INTERVAL_MS);

// Webhook endpoint
app.post('/ingest', async (req, res) => {
  const events = Array.isArray(req.body) ? req.body : [req.body];

  for (const event of events) {
    buffer.push({
      event_type: event.type ?? 'unknown',
      user_id: event.userId ?? 0,
      properties: JSON.stringify(event.properties ?? {}),
      created_at: new Date().toISOString().replace('T', ' ').slice(0, 19),
    });
  }

  if (buffer.length >= BATCH_SIZE) {
    await flushBuffer();
  }

  res.status(202).json({ queued: events.length, buffer_size: buffer.length });
});

Step 2: Choose a Server-Side or Managed Path

For high-volume streams, prefer a path that needs no application consumer:

• Kafka table engine — ClickHouse consumes a topic directly and a

materialized view pipes rows into your MergeTree table. No consumer to run.

• ClickPipes — ClickHouse Cloud's managed, code-free ingestion for Kafka,

Confluent, Amazon MSK, S3, and GCS.

• HTTP interface — bulk-load CSV / NDJSON / Parquet from files, remote

URLs, or S3 with plain curl, no client library.

Full DDL and configuration for all three: see

Ingestion methods.

Step 3: Make Ingestion Idempotent and Observable

Webhook retries and Kafka reprocessing deliver duplicates. Use a

ReplacingMergeTree keyed on a unique event_id so re-delivered events collapse

to one row, and query system.query_log to watch insert throughput and errors.

Full DDL, monitoring queries, and the batch-tuning matrix:

Deduplication & monitoring.

Output

Applying this skill produces:

• A running webhook receiver (POST /ingest) that buffers events and

batch-flushes to ClickHouse, returning 202 { queued, buffer_size }.

• Optionally, a Kafka engine table + materialized view (or a ClickPipes

pipe) that ingests a topic server-side with no application consumer.

• A ReplacingMergeTree dedup table keyed on event_id for idempotent,

retry-safe ingestion.

• Monitoring queries over system.query_log reporting inserts/minute,

rows, bytes, and insert exceptions in the last hour.

Error Handling

Examples

Ingest a webhook batch via the receiver (Step 1):

curl -X POST http://localhost:3000/ingest \
  -H 'Content-Type: application/json' \
  -d '[{"type":"signup","userId":42,"properties":{"plan":"pro"}}]'
# → 202 { "queued": 1, "buffer_size": 1 }

Bulk-load a Parquet file with no client (HTTP interface — see

Ingestion methods):

curl 'http://localhost:8123/?query=INSERT+INTO+analytics.events+FORMAT+Parquet' \
  --data-binary @events.parquet

Read deduplicated events (ReplacingMergeTree — see

Deduplication & monitoring):

SELECT * FROM analytics.events_dedup FINAL
WHERE created_at >= today() - 7;

Resources

• Ingestion methods — Kafka engine,

ClickPipes, HTTP bulk insert (full DDL)

• Deduplication & monitoring —

ReplacingMergeTree, system.query_log queries, best-practices matrix

• Kafka Integration
• ClickPipes
• HTTP Interface
• S3 Table Function

Next Steps

For query and server performance after ingestion is flowing, see

clickhouse-performance-tuning. For engine and schema choices on the target

table, see clickhouse-core-workflow-a.