clickhouse-webhooks-events

name	clickhouse-webhooks-events
description	Ingest data into ClickHouse from webhooks, Kafka, and streaming sources with batching, dedup, and exactly-once patterns. Use when building data ingestion pipelines, consuming webhook payloads, or integrating Kafka topics into ClickHouse. Trigger with "clickhouse ingestion", "clickhouse webhook", "clickhouse Kafka", "stream data to clickhouse", "clickhouse data pipeline".
allowed-tools	Read, Write, Edit, Bash(curl:*)
version	1.7.0
license	MIT
author	Jeremy Longshore <[e-mail]>
tags	["saas","database","analytics","clickhouse","olap"]
compatibility	Designed for Claude Code

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  () {
   (buffer. === ) ;
   batch = buffer.(, buffer.);

   {
     client.({
      : ,
      : batch,
      : ,
    });
    .();
  }  (err) {
    .(, (err  ).);
    buffer.(...batch);  
  }
}


(flushBuffer, );


app.(,  (req, res) => {
   events = .(req.) ? req. : [req.];

   ( event  events) {
    buffer.({
      : event. ?? ,
      : event. ?? ,
      : .(event. ?? {}),
      :  ().().(, ).(, ),
    });
  }

   (buffer. >= ) {
     ();
  }

  res.().({ : events., : buffer. });
});