The Complete Production SOP: How to Deploy and Run ClickHouse High-Throughput Batch Inserts in Autonomous AI Agents

---
name: "clickhouse-high-throughput-insert"
description: "Ingest and query billions of events with ClickHouse columnar database engine."
metadata:
  author: "ClickHouse"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "insert batch clickhouse"
  - "configure mergetree table"
  - "enable async insert clickhouse"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

# ClickHouse High-Throughput Batch Inserts

Optimizes ClickHouse data ingestion and storage: MergeTree table engines, primary key sorting keys, async inserts (async_insert=1), batching protocols, and partitioned data compression.

## When to Invoke
Invoke this skill whenever you need to:
- insert batch clickhouse
- configure mergetree table
- enable async insert clickhouse

## Execution SOP & Rules
1. Design tables using ENGINE = MergeTree() ORDER BY (tenant_id, event_time, user_id).
2. Partition by month or day: PARTITION BY toYYYYMM(event_time); avoid granular partitioning.
3. Batch inserts in chunks of 50,000 to 100,000 rows, or enable async_insert on the server.
4. Use low-cardinality types: LowCardinality(String) for columns with under 10,000 unique values.

## Verification Checklist
- Run local linting, syntax validation, and type-checks before reporting completion.
- Ensure all created or modified files adhere to workspace boundaries and avoid uncommitted secrets.
- Verify tests or build commands exit cleanly with return code 0.

import os
import clickhouse_connect

def get_high_throughput_client():
    return clickhouse_connect.get_client(
        host=os.getenv('CLICKHOUSE_HOST', 'localhost'),
        port=int(os.getenv('CLICKHOUSE_PORT', 8443)),
        username=os.getenv('CLICKHOUSE_USER', 'default'),
        password=os.getenv('CLICKHOUSE_PASSWORD', ''),
        database=os.getenv('CLICKHOUSE_DATABASE', 'telemetry'),
        secure=True,
        settings={
            'async_insert': 1,
            'wait_for_async_insert': 1,
            'async_insert_busy_timeout_ms': 200,
            'async_insert_max_data_size': 10485760
        }
    )

CREATE TABLE IF NOT EXISTS telemetry.events (
    tenant_id UInt32,
    event_time DateTime64(3, 'UTC'),
    user_id UUID,
    event_type LowCardinality(String),
    environment LowCardinality(String),
    payload String CODEC(ZSTD(3))
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (tenant_id, event_type, event_time, user_id)
SETTINGS index_granularity = 8192;

def test_batch_insertion(client):
    sample_data = [
        (101, '2026-03-30 12:00:00.000', 'a0eebc99-9c0b-4ef8-bb6d-6bb9bd380a11', 'click', 'prod', '{"btn": 1}')
    ] * 50000
    client.insert(
        table='events',
        data=sample_data,
        column_names=['tenant_id', 'event_time', 'user_id', 'event_type', 'environment', 'payload']
    )
    result = client.command('SELECT count() FROM telemetry.events')
    assert result >= 50000