Skip to main content

ClickHouse multi-table split

One pipeline can write to several ClickHouse tables, routing each record to the table for its kind — each table with its own columns, encoder, shard router, and part tuning. This is the connector-specific side of the multi-sink split concept: the split terminal is generic; here is how the pieces line up for ClickHouse.

Reach for it when your stream carries genuinely different record types (each wanting its own columns and ORDER BY) and you'd rather the ETL fan them out than a Null table + one MATERIALIZED VIEW per type. The trade-off — ETL-side fan-out CPU and part control vs. server-side fan-out — is measured in the multi-table split benchmark.

Configuration: a sinks: map​

Replace the single sink: section with a sinks: map keyed by name. Each entry is a full, ordinary ClickHouse sink section — its own table, columns, format, shards, and batch/inflight — so per-table part sizing is tuned independently. The map key is the name the chain resolves with ctx.sink(...).

sinks:
  gauge:
    clickhouse:
      table: metrics_gauge
      columns: [host, ts_ms, name, value]   # this table's own columns/order
      format: native
      shards:
        - replicas: ["${CLICKHOUSE_URL:-http://localhost:8123}"]
      batch: { max_rows: 500000, linger: 1s }

  text:
    clickhouse:
      table: metrics_text
      columns: [host, ts_ms, name, text]    # a different schema
      format: native
      shards:
        - replicas: ["${CLICKHOUSE_URL:-http://localhost:8123}"]
      # A low-volume table: linger longer to build bigger parts (at the cost of
      # checkpoint lag — see the tuning note below).
      batch: { max_rows: 500000, linger: 5s }

sink: and sinks: are mutually exclusive; a single-sink pipeline keeps using sink: (addressed as "default"). See the configuration reference. Every key documented for a single ClickHouse sink applies per-entry.

Wiring: one sink, encoder, and router per table​

Each table is an ordinary ClickHouseSink — build it from its named config section, mint its Native encoder (from its system.columns) and its shard router, then install it with add_sink(name, ...). This is exactly the single-sink wiring, N times:

// One sink per table, from the `sinks:` map.
let gauge_sink = etl::clickhouse::config::from_component_config(pipeline.config().sink_config("gauge")?)?;
let text_sink  = etl::clickhouse::config::from_component_config(pipeline.config().sink_config("text")?)?;

// Each table's own columnar encoder (its Native schema) and shard router.
let gauge_router = gauge_sink.router::<GaugeFam>(host_key);
let text_router  = text_sink.router::<TextFam>(host_key);
let gauge_enc = NativeEncoder::<GaugeFam>::new(pipeline.block_on(gauge_sink.native_schema())?);
let text_enc  = NativeEncoder::<TextFam>::new(pipeline.block_on(text_sink.native_schema())?);

let report = pipeline
    .add_sink("gauge", gauge_sink)?
    .add_sink("text",  text_sink)?
    .chains(move |ctx| {
        let mut split = chain::<SampleFam, _>(deserializer.clone())
            .with_metrics(ctx.pipeline.clone(), "main")
            .split(ChunkConfig::default(), ErrorPolicy::Skip);
        let gauge = split.add::<GaugeFam, _, _>(gauge_enc.clone(), gauge_router.clone(), ctx.sink("gauge"));
        let text  = split.add::<TextFam,  _, _>(text_enc.clone(),  text_router.clone(),  ctx.sink("text"));
        split.route(move |s: Sample<'_>, out| match s.kind {
            "gauge" => out.emit(gauge, GaugeRow { /* … */ }),
            "text"  => out.emit(text,  TextRow  { /* … */ }),
            _       => {}
        }).build()
    })
    .run(source)?;

The complete, compiling program is crates/etl/examples/multi_table_split.rs (--features full,avro-fast). See the concept page for the routing DSL walkthrough.

What carries over per table, unchanged​

Each branch is a full ClickHouse sink, so everything on the ClickHouse sink page holds per table:

  • Native (or RowBinary) encoding on pipeline threads, positional against that table's columns. Mixed formats across tables are fine — each sink chooses its own.
  • Direct-to-shard writes with wait_end_of_query=1; a table's write is its own durable-ack point. The split holds the source batch until every table it touched has acked (see delivery).
  • Per-batch insert_deduplication_token per table. The same window warning applies: set non_replicated_deduplication_window on every target table, or retries duplicate silently.
  • Replica rotation, circuit breaker, readiness probe per table. /readyz reports connected only when every table's replicas probe healthy.
  • Metrics per table: each sink's etl_sink_* series carry the sink name as their component label, so etl_sink_records_total{component="gauge"} and {component="text"} are distinct.

Per-table part tuning — and the checkpoint-lag cost​

The whole point of splitting in the ETL is per-table part control: size each table's batch.max_rows/max_bytes/linger for its volume. A hot table seals on rows in milliseconds; a rare table would trickle, so you raise its linger to build bigger parts.

[!IMPORTANT] linger on a rare table is not free: because the source watermark is held until every table a batch touched has flushed, a long rare-table linger holds the offsets of every source batch that carried even one rare record — inflating checkpoint lag and restart replay. Keep rare-table linger well under checkpoint.stalled_fail_after, and prefer sealing on a modest max_rows so linger only bites in genuine lulls. The full reasoning — including segregating the source by kind to dissolve the tension — is in the concept page's trade-off section.

distributed_check per table​

Each table can carry its own distributed_check: the router placement is verified against that table's Distributed DDL independently, so a sharded multi-table deployment keeps every table aligned with its cluster.