clickhouse-kafka-pipeline.sql

-- Kafka (Redpanda) -> ClickHouse pipeline for crawl-url-events.
-- A Kafka engine table is PASSIVE; a Materialized View drives it into a MergeTree target:
--   crawl_frontier_queue (Kafka) --[crawl_frontier_mv]--> crawl_frontier_events (ReplicatedMergeTree)
--
-- Run: oc -n ddg-index-clickhouse exec -i chi-main-ddgindex-0-0-0 -c clickhouse -- \
--        clickhouse-client --multiquery < clickhouse-kafka-pipeline.sql
--
-- NOTE on offsets: this ClickHouse rejects `kafka_auto_offset_reset` as a table setting, and
-- the default is EARLIEST -> attaching the MV backfills the whole topic (~250M rows). Pick one:
--   (A) runtime, no config change -- seek the consumer group to the end BEFORE creating the MV:
--         oc -n ddg-redpanda-prod exec redpanda-0 -c redpanda -- \
--           rpk group seek crawl-frontier-clickhouse-ddgindex --to end --topics crawl-url-events
--       ClickHouse only falls back to auto.offset.reset when the group has NO committed offsets,
--       so seeding it to 'end' makes it start from new messages only.
--   (B) persistent -- set it in the CHI <kafka> config (operator), applies to all Kafka tables:
--         spec.configuration.settings += { 'kafka/auto_offset_reset': 'latest' }   (needs apply)
--   For a full historical backfill instead, do neither and let it default to earliest.

CREATE DATABASE IF NOT EXISTS ddgindex ON CLUSTER 'ddgindex';

-- 1) Kafka source (single-line broker list; NO kafka_auto_offset_reset -- unsupported here).
CREATE TABLE IF NOT EXISTS ddgindex.crawl_frontier_queue ON CLUSTER 'ddgindex'
(
    schema_version UInt16, event_id String, event_type LowCardinality(String),
    event_time DateTime64(3, 'UTC'), raw_url String, normalized_url String,
    raw_hash String, norm_hash String, normalization_version LowCardinality(String),
    registered_domain LowCardinality(String), source LowCardinality(String),
    request_id String, trace_id String, submission_event_id String,
    request_endpoint LowCardinality(String), max_content_age_seconds Nullable(UInt32),
    dispatch_id String, attempt_id String, attempt_kind LowCardinality(String),
    retry_attempt_count UInt32, rate_limit_key String,
    priority_version LowCardinality(String), priority_components Map(String, Float64),
    budget_rps Nullable(Float64), lease_until Nullable(DateTime64(3, 'UTC')),
    result LowCardinality(String), reason LowCardinality(String),
    http_status Nullable(UInt16), retry_at Nullable(DateTime64(3, 'UTC')),
    blob_container LowCardinality(String), content_hash String,
    content_quality_version LowCardinality(String), content_quality_label LowCardinality(String),
    content_quality_additional_label LowCardinality(String), user_agent LowCardinality(String),
    compared_content_hash String, content_improved Nullable(Int8)
)
ENGINE = Kafka
SETTINGS
    kafka_broker_list = 'redpanda-0.prod.use.ddgindex.duckduckgo.com:31092,redpanda-1.prod.use.ddgindex.duckduckgo.com:31092,redpanda-2.prod.use.ddgindex.duckduckgo.com:31092,redpanda-3.prod.use.ddgindex.duckduckgo.com:31092,redpanda-4.prod.use.ddgindex.duckduckgo.com:31092,redpanda-5.prod.use.ddgindex.duckduckgo.com:31092,redpanda-6.prod.use.ddgindex.duckduckgo.com:31092',
    kafka_topic_list = 'crawl-url-events',
    kafka_group_name = 'crawl-frontier-clickhouse-ddgindex',
    kafka_format = 'JSONEachRow',
    kafka_security_protocol = 'ssl',
    kafka_handle_error_mode = 'stream',
    kafka_max_block_size = 100000,
    date_time_input_format = 'best_effort',
    input_format_skip_unknown_fields = 1;

-- 2) Target (37 columns copied from the Kafka table), replicated across both replicas.
CREATE TABLE IF NOT EXISTS ddgindex.crawl_frontier_events ON CLUSTER 'ddgindex'
AS ddgindex.crawl_frontier_queue
ENGINE = ReplicatedMergeTree
PARTITION BY toYYYYMM(event_time)
ORDER BY (event_type, registered_domain, event_time, event_id);

-- 3) Materialized View -- creating THIS is what starts consumption. Do your offset choice
--    (A or B above) FIRST, then create it.
CREATE MATERIALIZED VIEW IF NOT EXISTS ddgindex.crawl_frontier_mv ON CLUSTER 'ddgindex'
TO ddgindex.crawl_frontier_events
AS SELECT * FROM ddgindex.crawl_frontier_queue;

-- 4) (Optional) capture unparseable messages (kafka_handle_error_mode='stream').
CREATE TABLE IF NOT EXISTS ddgindex.crawl_frontier_errors ON CLUSTER 'ddgindex'
(error String, raw_message String, ingested_at DateTime DEFAULT now())
ENGINE = ReplicatedMergeTree ORDER BY ingested_at;
CREATE MATERIALIZED VIEW IF NOT EXISTS ddgindex.crawl_frontier_errors_mv ON CLUSTER 'ddgindex'
TO ddgindex.crawl_frontier_errors
AS SELECT _error AS error, _raw_message AS raw_message
   FROM ddgindex.crawl_frontier_queue WHERE length(_error) > 0;

-- Verify (after the MV exists + messages flow):
--   SELECT count() FROM ddgindex.crawl_frontier_events;
--   SELECT database, table, num_messages_read, assignments.topic, assignments.current_offset
--     FROM system.kafka_consumers WHERE table='crawl_frontier_queue' FORMAT Vertical;