Clickhouse Webhooks Events

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.

Source: jeremylongshore/claude-code-plugins-plus-skills → plugins/saas-packs/clickhouse-pack/skills/clickhouse-webhooks-events/SKILL.md