Skip to main content

Your first pipeline: Kafka → Avro → ClickHouse

This tutorial builds the flagship shape: consume Confluent-framed Avro from Kafka, validate and transform records in an operator chain, and write them to ClickHouse in large batches — at-least-once, drained gracefully on SIGTERM. It mirrors crates/etl/examples/kafka_avro_to_clickhouse.rs and its YAML (crates/etl/examples/kafka_avro_to_clickhouse.yaml); every snippet below is lifted from those files.

You need Kafka, a Confluent-compatible schema registry, and ClickHouse, plus the full feature (see Installation). From the repository:

cargo run --release -p etl --example kafka_avro_to_clickhouse --features full

1. Create the target table​

Before running anything, create the table — and note the deduplication window setting:

CREATE TABLE orders (
    id           UInt64,
    customer     String,
    amount_cents Int64,
    ts_ms        Int64
) ENGINE = MergeTree ORDER BY id
  SETTINGS non_replicated_deduplication_window = 100;

[!WARNING] On plain MergeTree, insert deduplication silently no-ops unless the table sets non_replicated_deduplication_window greater than 0 (the server default is 0). Without it, the sink's dedup tokens do nothing and a retried batch lands twice. Replicated*MergeTree tables default to a window of 100 and don't need the setting. Even with the window, dedup tokens only cover same-batch retries — not crash replay. Read Delivery guarantees before choosing a table engine for production.

2. The record type​

One struct travels end to end. Deserialize reads it from Avro (field names match the writer schema); Serialize writes it as RowBinary:

use serde::{Deserialize, Serialize};

#[derive(Clone, Debug, Deserialize, Serialize)]
struct Order {
    id: u64,
    customer: String,
    amount_cents: i64,
    ts_ms: i64,
}

[!IMPORTANT] RowBinary carries no column names — the struct's field declaration order must match the columns list in the YAML. Order is the wire contract. The validate_schema option below catches mismatches against the live table at startup instead of at 3 a.m.

3. The YAML​

Tuning lives in YAML next to the binary; the operator graph never does. Every ${VAR:-default} interpolates from the environment (in Kubernetes: ConfigMap/Secret env). Framework sections (pipeline, checkpoint, backpressure, metrics) are typed and validated; the source, deserializer, and sink bodies belong to their connectors:

pipeline:
  name: orders
  # threads: 4          # default: available cores minus the I/O reserve
  io_threads: 2

checkpoint:
  interval: 5s
  max_pending_batches: 1024
  drain_timeout: 25s    # keep below terminationGracePeriodSeconds

backpressure:
  # Sizing rule (docs/DESIGN.md § Backpressure): the budget's low watermark
  # must hold ~2x (shards x inflight x batch.max_bytes + queued chunks).
  max_inflight_bytes: 1GiB

metrics:
  exporter: prometheus
  listen: 0.0.0.0:9090  # /metrics, /healthz, /readyz

source:
  kafka:
    brokers: ${KAFKA_BROKERS:-localhost:9092}
    topic: ${KAFKA_TOPIC:-orders}
    group_id: ${KAFKA_GROUP:-orders-etl}
    commit_interval: 5s
    rdkafka:
      # Raw librdkafka passthrough. Properties the framework owns for
      # correctness (offset storage, auto-commit) are rejected here.
      auto.offset.reset: earliest

deserializer:
  avro:
    mode: confluent
    registry:
      url: ${SCHEMA_REGISTRY_URL:-http://localhost:8081}
    prewarm_subjects: ["${KAFKA_TOPIC:-orders}-value"]

sink:
  clickhouse:
    table: ${CLICKHOUSE_TABLE:-orders}
    # Column order is the RowBinary wire contract: it must match the row
    # struct's field declaration order.
    columns: [id, customer, amount_cents, ts_ms]
    shards:
      - replicas: ["${CLICKHOUSE_URL:-http://localhost:8123}"]
    user: ${CLICKHOUSE_USER:-default}
    password: ${CLICKHOUSE_PASSWORD:-}
    # Fail startup (and the first record) when the columns, row struct,
    # and live table disagree: off | names | full.
    validate_schema: full
    batch:
      max_rows: 500000
      max_bytes: 128MiB
      linger: 1s
    inflight:
      max_per_shard: 2

The backpressure comment is load-bearing: with a saturated source, an undersized budget collapses throughput. The rule and the arithmetic are in Backpressure.

4. The binary​

The builder owns the process plumbing (telemetry, metrics exporter, the shared I/O runtime, shard queues, sink workers, probes). The code you write is only what is genuinely this pipeline's: connector construction, schema validation, and the chain.

use etl::avro::AvroDeserializerBuilder;
use etl::clickhouse::ClickHouseEncoder;
use etl::kafka::KafkaSource;
use etl::prelude::*;
use std::path::Path;

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let config_path = std::env::var("ETL_CONFIG")
        .unwrap_or_else(|_| "pipeline.yaml".to_string());
    let pipeline = Pipeline::from_path(Path::new(&config_path))?;

    // Source: one Kafka consumer per process; partitions become lanes
    // fanned across pipeline threads.
    let source = KafkaSource::from_component_config(&pipeline.config().source)?;

    // Deserializer: Confluent-framed Avro. Schemas arrive via an async
    // fetcher on the I/O runtime; a cache miss never blocks a pipeline
    // thread — the batch retries once the schema lands.
    let deser_section = pipeline
        .config()
        .deserializer
        .as_ref()
        .ok_or("this pipeline requires a `deserializer` section")?;
    let deserializer =
        AvroDeserializerBuilder::from_component(deser_section, &pipeline.io_handle())?
            .build_serde::<Order>()?;

    // Sink: the connector turns its YAML section into everything the
    // builder needs — writer, per-shard replica endpoints, pool tuning,
    // readiness probe.
    let sink = etl::clickhouse::config::from_component_config(&pipeline.config().sink)?;

    // Opt-in fail-fast schema validation (validate_schema: names|full):
    // checks the configured columns against every replica's live table
    // NOW, before any thread spawns.
    let encoder = match pipeline.block_on(sink.validate_schema())? {
        Some(schema) => ClickHouseEncoder::<Owned<Order>>::with_schema(schema),
        None => ClickHouseEncoder::<Owned<Order>>::new(),
    };

    // One identical chain per pipeline thread, fully monomorphized.
    let report = pipeline
        .sink(sink)?
        .chains(move |ctx| {
            chain_owned::<Order, _>(deserializer.clone())
                .with_metrics(ctx.pipeline, "main")
                .try_map(
                    |order: Order| {
                        if order.amount_cents >= 0 {
                            Ok(order)
                        } else {
                            Err("negative amount")
                        }
                    },
                    ErrorPolicy::Skip,
                )
                .sink(
                    encoder.clone(),
                    KeyHashRouter,
                    ChunkConfig::default(),
                    ctx.queues,
                    ctx.budget,
                )
                .build()
        })
        .run(source)?;

    report.log();
    std::process::exit(report.exit_code());
}

Walking the assembly:

  • Pipeline::from_path loads the YAML and owns init: JSON logs (RUST_LOG overrides the filter; call etl::telemetry::init first to customize), the metrics exporter — installed before any handle can exist — and the shared I/O runtime. Point ETL_CONFIG elsewhere to reconfigure without recompiling.
  • pipeline.io_handle() / pipeline.block_on(..) give connectors the I/O runtime before run — the schema-registry fetcher and the validate_schema pre-flight both use it.
  • .sink(sink) spawns the sharded sink workers and wires the drain and the /readyz probe.
  • .chains(..) installs the per-thread chain factory. The try_map with ErrorPolicy::Skip drops bad records, counts them on etl_operator_records_dropped_total, and keeps going — see Error handling.
  • .run(source) blocks until SIGTERM/SIGINT (drain) or a fatal error, then returns an exit report.

Each step is a thin composition of public primitives you can drop down to — the full mapping is in the etl::pipeline::Pipeline module docs and in Manual assembly.

5. Run it, watch it, stop it​

export KAFKA_BROKERS=localhost:9092
export SCHEMA_REGISTRY_URL=http://localhost:8081
export CLICKHOUSE_URL=http://localhost:8123
cargo run --release -p etl --example kafka_avro_to_clickhouse --features full

Probes and metrics are on the admin listener from the YAML:

curl localhost:9090/readyz    # assignment received and sink connected
curl localhost:9090/healthz   # poll loop alive, watermarks moving
curl localhost:9090/metrics   # Prometheus scrape endpoint

The SIGTERM story. Send the process SIGTERM (what Kubernetes does on pod termination) and it drains rather than dies: lanes stop polling, chains flush, sink batches complete — bounded by checkpoint.drain_timeout — and offsets commit synchronously before the process exits. If the sink is down at the deadline, unflushed batches are abandoned loudly (metric + log) and replay on restart: at-least-once holds either way. Keep drain_timeout below your pod's terminationGracePeriodSeconds so the drain finishes before the SIGKILL. Details in Graceful shutdown.

Where next​