ClickHouse Data Ingestion Skill

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

| Error | Cause | Solution | |-------|-------|----------| | Too many parts | Single-row inserts | Batch inserts (10K+ rows) | | Cannot parse input | Wrong format | Match format to data structure | | TIMEOUT on large insert | Slow network | Enable compression, split batch | | Duplicate events | Webhook retries | Use ReplacingMergeTree + event_id |

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.

Agent Skills: ClickHouse Data Ingestion

Install this agent skill to your local

Skill Files