Kafka 引擎与实时管道

CREATE TABLE kafka_events_queue (
    user_id UInt64,
    event_type String,
    value Float64,
    event_time DateTime
) ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'kafka1:9092,kafka2:9092',
    kafka_topic_list = 'events',
    kafka_group_name = 'ch_events_consumer',
    kafka_format = 'JSONEachRow',
    kafka_num_consumers = 4,
    kafka_max_block_size = 65536;

-- 直接消费一次（消费后 offset 前移，谨慎使用）
SELECT count()
FROM kafka(
    'kafka1:9092', 'events', 'ad_hoc_group',
    'JSONEachRow',
    'user_id UInt64, event_type String'
);

-- 目标表：持久化明细
CREATE TABLE events (
    event_time DateTime,
    user_id UInt64,
    event_type String,
    page_url String,
    value Float64
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (event_time, user_id);

-- 物化视图：Kafka 消息 → 目标表
CREATE MATERIALIZED VIEW events_kafka_mv
TO events
AS SELECT
    event_time,
    user_id,
    event_type,
    page_url,
    value
FROM kafka_events_queue;

{"user_id": 1, "event_type": "view",
 "page": {"url": "/home", "ref": "https://x.com"},
 "tags": ["a", "b"]}

CREATE TABLE kafka_queue_nested (
    user_id UInt64,
    event_type String,
    page_url String,
    ref String,
    tags Array(String)
) ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'kafka:9092',
    kafka_topic_list = 'events_nested',
    kafka_group_name = 'ch_nested',
    kafka_format = 'JSONEachRow';

CREATE MATERIALIZED VIEW nested_mv TO events_nested AS
SELECT
    user_id,
    event_type,
    page.url AS page_url,
    page.ref AS ref,
    tags
FROM kafka_queue_nested;

-- 原始消息先落库再后续解析
CREATE TABLE kafka_raw (
    raw String
) ENGINE = Kafka()
SETTINGS kafka_broker_list='kafka:9092',
         kafka_topic_list='events_raw',
         kafka_group_name='ch_raw',
         kafka_format='JSONAsString';

-- 排查：直接看 Kafka 消费组的 lag
-- 用 Kafka 自带的命令行（在 Kafka 机器上执行）
kafka-consumer-groups.sh --bootstrap-server kafka1:9092 \
    --describe --group ch_events_consumer

CREATE TABLE events_dedup (
    event_time DateTime,
    user_id UInt64,
    event_type String,
    msg_id String,          -- 消息唯一 ID（如 Kafka offset 或业务 ID）
    value Float64
) ENGINE = ReplacingMergeTree(msg_id)
PARTITION BY toYYYYMM(event_time)
ORDER BY (user_id, event_time);

CREATE TABLE kafka_events_queue (
    user_id UInt64,
    event_type String
) ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'kafka:9092',
    kafka_topic_list = 'events',
    kafka_group_name = 'ch_events',
    kafka_format = 'JSONEachRow',
    kafka_num_consumers = 8,        -- 每节点消费者线程
    kafka_thread_per_consumer = 0;  -- 每消费者线程数（0 = 自动）

-- 会话级：让目标表写入也并行
SET max_insert_threads = 4;
SET async_insert = 1;

-- 订阅多个主题
CREATE TABLE kafka_multi (
    topic String,
    user_id UInt64,
    event_type String,
    event_time DateTime
) ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'kafka:9092',
    kafka_topic_list = 'page_views,clicks,purchases',  -- 逗号分隔多主题
    kafka_group_name = 'ch_multi',
    kafka_format = 'JSONEachRow';

CREATE MATERIALIZED VIEW views_mv TO page_views AS
SELECT * FROM kafka_multi WHERE topic = 'page_views';

CREATE TABLE events_by_day (
    event_date Date,
    hour UInt8,
    user_id UInt64,
    event_type String
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_date)
ORDER BY (event_date, user_id);

CREATE MATERIALIZED VIEW kafka_day_mv TO events_by_day AS
SELECT
    toDate(event_time) AS event_date,
    toHour(event_time) AS hour,
    user_id,
    event_type
FROM kafka_events_queue;

-- 动态分区管理：批量建未来分区
SELECT 'ALTER TABLE events_by_day ADD PARTITION ' || toString(toYYYYMM(now() + INTERVAL 1 MONTH));

// Go 直写参考：攒批 1000 行或 5 秒一次
rows := make([][]interface{}, 0, 1000)
for msg := range consumer.Messages() {
    rows = append(rows, parse(msg))
    if len(rows) >= 1000 {
        ch.Exec(ctx, "INSERT INTO events VALUES", rows)
        rows = rows[:0]
    }
}

CREATE TABLE kafka_events_queue (...) ENGINE = Kafka()
SETTINGS
    ...,
    kafka_skip_broken_messages = 1000;   -- 跳过解析失败的消息（每 Block 上限）

-- 目标表写入是否持续
SELECT max(event_time) AS last_ts, count() AS rows_last_block
FROM events;

-- 目标表 parts 增长（反映消费与合并节奏）
SELECT count() AS parts, sum(rows) AS rows
FROM system.parts WHERE table = 'events' AND active = 1;