Imported from jeremylongshore/tons-of-skills-marketplace (skills/.curated/clickhouse-webhooks-events/SKILL.md). Install upstream with npx skills add jeremylongshore/tons-of-skills-marketplace --skill clickhouse-webhooks-events. Copyright stays with the author (MIT).

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.