ClickHouse 进阶实战#

1
-- ============================================================
2
-- 1. 创建 Kafka 引擎表（接入表）
3
--    作用：作为数据管道，从 Kafka 拉取消息，本身不存储数据。
4
--    注意：该表仅用于消费，查询它只能看到极少量未转存的缓存数据。
5
-- ============================================================
6
CREATE TABLE analytics.ods_store_events_kafka
7
(
8
    event_id UUID,
9
    tenant_id String,
10
    store_id String,
11
    member_id Nullable(UInt64),
12
    device_id Nullable(String),
13
    city String,
14
    event_name String,
15
    order_id Nullable(UInt64),
16
    product_id Nullable(UInt64),
17
    page_code String,
18
    event_time DateTime64(3, 'Asia/Shanghai'),
19
    properties String
20
)
21
ENGINE = Kafka
22
SETTINGS
23
    kafka_broker_list = 'localhost:9092',          -- Kafka 集群地址
24
    kafka_topic_list = 'test-topic',               -- 订阅的 Topic
25
    kafka_group_name = 'clickhouse-local-consumer',-- 消费者组名（保证偏移量管理）
26
    kafka_format = 'JSONEachRow',                  -- 消息格式（每行一个 JSON）
27
    kafka_num_consumers = 1,                       -- 消费者线程数（可根据分区数调整）
28
    kafka_handle_error_mode = 'stream';
29

30
-- ============================================================
31
-- 2. 创建物化视图（数据管道核心）
32
--    作用：自动将 Kafka 接入表的数据转换并写入目标 MergeTree 表。
33
--    特点：视图本身不存储数据，每次 Kafka 新消息到达时触发写入。
34
-- ============================================================
35
CREATE MATERIALIZED VIEW analytics.mv_kafka_to_store_events
36
TO analytics.store_events
37
AS SELECT
38
    event_id, tenant_id, store_id, member_id, device_id, city,
39
    event_name, order_id, product_id, page_code, event_time, properties
40
FROM analytics.ods_store_events_kafka;   -- 数据来源为 Kafka 引擎表

1
-- ============================================================
2
-- 3. 验证查询（查询目标 MergeTree 表，而非 Kafka 表）
3
--    目的：确认数据已成功写入，并查看最近事件。
4
-- ============================================================
5
SELECT event_time, store_id, event_name, order_id, ingest_time
6
FROM analytics.store_events
7
WHERE tenant_id = 'tenant-a'
8
ORDER BY ingest_time DESC
9
LIMIT 20;

1
   ┌──────────────event_time─┬─store_id─────┬─event_name──────┬─order_id─┬─────────ingest_time─┐
2
1. │ 2026-08-25 12:30:00.000 │ store-hz-001 │ page_view       │     ᴺᵁᴸᴸ │ 2026-08-25 13:51:38 │
3
2. │ 2026-07-21 10:00:01.123 │ store-hz-001 │ entry_detected  │     ᴺᵁᴸᴸ │ 2026-07-24 11:34:59 │
4
3. │ 2026-07-21 10:01:03.456 │ store-hz-001 │ product_view    │     ᴺᵁᴸᴸ │ 2026-07-24 11:34:59 │
5
4. │ 2026-07-21 10:02:20.000 │ store-hz-001 │ payment_success │   200001 │ 2026-07-24 11:34:59 │
6
   └─────────────────────────┴──────────────┴─────────────────┴──────────┴─────────────────────┘

1
普通查询：明细表 → 每次查询现场聚合 → 返回
2
物化视图：明细写入 → 增量聚合 → 结果表 → 查询直接读结果

1
CREATE TABLE analytics.store_event_1h
2
(
3
    hour DateTime,
4
    tenant_id LowCardinality(String),
5
    store_id LowCardinality(String),
6
    event_name LowCardinality(String),
7
    event_count AggregateFunction(count),
8
    member_uv AggregateFunction(uniqCombined64, Nullable(UInt64))
9
)
10
ENGINE = AggregatingMergeTree
11
PARTITION BY toYYYYMM(hour)
12
ORDER BY (tenant_id, store_id, hour, event_name);
13

14
CREATE MATERIALIZED VIEW analytics.mv_store_events_to_1h
15
TO analytics.store_event_1h
16
AS SELECT
17
    toStartOfHour(event_time) AS hour,
18
    tenant_id, store_id, event_name,
19
    countState() AS event_count,
20
    uniqCombined64State(member_id) AS member_uv
21
FROM analytics.store_events
22
GROUP BY hour, tenant_id, store_id, event_name;

1
INSERT INTO analytics.store_events
2
    (event_id, tenant_id, store_id, member_id, device_id, city, event_name,
3
     order_id, product_id, page_code, event_time, properties)
4
SELECT
5
    generateUUIDv4(),
6
    'tenant-a',
7
    concat('store-', toString(number % 100)),  -- 生成 100 个不同的门店
8
    toUInt64(number % 100000),
9
    concat('pos-', toString(number % 20)),
10
    '杭州',
11
    if(number % 10 = 0, 'payment_success',
12
       if(number % 10 = 1, 'payment_failed', 'entry_detected')),
13
    if(number % 10 IN (0, 1), toUInt64(number), NULL),
14
    if(number % 10 IN (0, 1), toUInt64(number % 1000), NULL),
15
    'checkout',
16
    toDateTime64('2026-07-21 00:00:00', 3) + toIntervalSecond(number % 86400),
17
    '{}'
18
FROM numbers(100000);

1
SELECT toStartOfHour(event_time) AS hour,
2
       countIf(event_name = 'entry_detected') AS entry_count
3
FROM analytics.store_events
4
WHERE tenant_id = 'tenant-a'
5
  AND event_date >= '2026-07-01'
6
  AND event_date < '2026-08-01'
7
GROUP BY hour;
8

9
-- 24 行已获取 - 0.013s, 2026-08-25 17:55:10

1
SELECT hour,
2
       countMerge(event_count) AS entry_count
3
FROM analytics.store_event_1h
4
WHERE tenant_id = 'tenant-a'
5
  AND event_name = 'entry_detected'
6
  AND hour >= '2026-07-01 00:00:00'
7
  AND hour < '2026-08-01 00:00:00'
8
GROUP BY hour;
9

10
-- 24 行已获取 - 0.01s, 2026-08-25 17:56:47

1
10:00  顾客支付成功
2
10:08  设备恢复网络，消息进入 ClickHouse
3
10:20  顾客退款
4
次日    按最终订单快照生成财务日报

1
INSERT INTO analytics.store_events
2
    (event_id, tenant_id, store_id, member_id, device_id, city, event_name,
3
     order_id, product_id, page_code, event_time, properties)
4
SELECT
5
    generateUUIDv4(),
6
    concat('tenant-', toString(number % 3)),                          -- 3 个租户
7
    concat('store-', toString(number % 200)),                         -- 200 个门店
8
    toUInt64(number % 100000),                                       -- 会员 ID 重复出现
9
    concat('pos-', toString(number % 50)),                           -- 50 个设备
10
    if(number % 10 < 3, '杭州', if(number % 10 < 6, '上海', '北京')), -- 3 个城市
11
    if(number % 10 = 0, 'payment_success',
12
       if(number % 10 = 1, 'payment_failed', 'entry_detected')),     -- 事件类型
13
    if(number % 10 IN (0, 1), toUInt64(number), NULL),               -- order_id 仅支付事件有值
14
    if(number % 10 IN (0, 1), toUInt64(number % 2000), NULL),        -- product_id
15
    'checkout',
16
    toDateTime64('2026-07-01 00:00:00', 3) + toIntervalSecond(number % 259200), -- 分散在 3 天（72 小时）
17
    '{}'
18
FROM numbers(5000000);  -- ← 改这个数字调整行数，例如 5000000 为 500 万

1
ORDER BY (tenant_id, store_id, event_date, event_time, event_name)

1
SELECT *
2
FROM analytics.store_events
3
WHERE toDate(event_time) >= '2026-07-01'
4
  AND toDate(event_time) < '2026-08-01'
5
  AND event_name = 'payment_success';

1
SELECT store_id, count() AS payment_count
2
FROM analytics.store_events
3
WHERE tenant_id = 'tenant-a'
4
  AND event_date >= '2026-07-01'
5
  AND event_date < '2026-08-01'
6
  AND event_name = 'payment_success'
7
GROUP BY store_id
8
ORDER BY payment_count DESC;

1
-- 1. 添加索引定义
2
ALTER TABLE analytics.store_events
3
ADD INDEX idx_product_id product_id TYPE bloom_filter(0.01) GRANULARITY 4;
4

5
-- 2. 物化索引（构建数据，这一步可能会消耗一点时间，但你的 10 万行数据几秒内完成）
6
ALTER TABLE analytics.store_events
7
MATERIALIZE INDEX idx_product_id;

1
SELECT
2
    name,
3
    type,
4
    granularity,
5
    formatReadableSize(index_size) AS index_size
6
FROM system.data_skipping_indices
7
WHERE table = 'store_events';

1
SELECT
2
    query_duration_ms, read_rows,
3
    formatReadableSize(read_bytes) AS read_size, query
4
FROM system.query_log
5
WHERE type = 'QueryFinish'
6
  AND event_time >= now() - INTERVAL 1 HOUR
7
ORDER BY read_rows DESC
8
LIMIT 10;

1
SELECT
2
    partition,
3
    count() AS active_parts,
4
    sum(rows) AS rows,
5
    formatReadableSize(sum(bytes_on_disk)) AS disk_size
6
FROM system.parts
7
WHERE database = 'analytics'
8
  AND table = 'store_events'
9
  AND active
10
GROUP BY partition
11
ORDER BY active_parts DESC;

1
CREATE TABLE analytics.api_access_log
2
(
3
    request_id UUID,
4
    request_time DateTime,
5
    service_name LowCardinality(String),
6
    status_code UInt16,
7
    cost_ms UInt32,
8
    message String
9
)
10
ENGINE = MergeTree
11
PARTITION BY toYYYYMM(request_time)
12
ORDER BY (service_name, request_time)
13
TTL request_time + INTERVAL 90 DAY DELETE;

1
INSERT INTO analytics.store_events
2
    (event_id, tenant_id, store_id, member_id, device_id, city, event_name,
3
     order_id, product_id, page_code, event_time, properties)
4
VALUES
5
    (generateUUIDv4(), 'tenant-a', 'store-hz-001', 10001, 'pos-01', '杭州', 'entry_detected', NULL, NULL, 'store_entry', '2026-07-21 10:00:01', '{}'),
6
    (generateUUIDv4(), 'tenant-a', 'store-hz-001', 10001, 'pos-01', '杭州', 'payment_success', 200001, 1001, 'checkout', '2026-07-21 10:02:20', '{"software_version":"2.4.1","cost_ms":680}'),
7
    (generateUUIDv4(), 'tenant-a', 'store-hz-001', 10002, 'pos-01', '杭州', 'entry_detected', NULL, NULL, 'store_entry', '2026-07-21 10:05:01', '{}'),
8
    (generateUUIDv4(), 'tenant-a', 'store-hz-001', 10002, 'pos-01', '杭州', 'payment_success', 200002, 1002, 'checkout', '2026-07-21 10:06:20', '{"software_version":"2.4.1","cost_ms":720}'),
9
    (generateUUIDv4(), 'tenant-a', 'store-hz-001', 10003, 'pos-02', '杭州', 'entry_detected', NULL, NULL, 'store_entry', '2026-07-21 10:10:01', '{}'),
10
    (generateUUIDv4(), 'tenant-a', 'store-hz-001', 10003, 'pos-02', '杭州', 'payment_failed', 200003, 1003, 'checkout', '2026-07-21 10:11:20', '{"software_version":"2.4.2","error_code":"PAY_TIMEOUT","cost_ms":3200}'),
11
    (generateUUIDv4(), 'tenant-a', 'store-hz-002', 10011, 'pos-03', '宁波', 'entry_detected', NULL, NULL, 'store_entry', '2026-07-21 10:03:01', '{}'),
12
    (generateUUIDv4(), 'tenant-a', 'store-hz-002', 10011, 'pos-03', '宁波', 'payment_failed', 300001, 1001, 'checkout', '2026-07-21 10:04:20', '{"software_version":"2.4.2","error_code":"PAY_TIMEOUT","cost_ms":4100}'),
13
    (generateUUIDv4(), 'tenant-a', 'store-hz-002', 10012, 'pos-03', '宁波', 'entry_detected', NULL, NULL, 'store_entry', '2026-07-21 10:08:01', '{}'),
14
    (generateUUIDv4(), 'tenant-a', 'store-hz-002', 10012, 'pos-03', '宁波', 'payment_failed', 300002, 1002, 'checkout', '2026-07-21 10:09:20', '{"software_version":"2.4.2","error_code":"PAY_TIMEOUT","cost_ms":3800}');

1
-- ============================================================
2
-- 查询各门店在指定日期（2026-07-21）的转化漏斗核心指标
3
-- 数据来源：analytics.store_events（门店事件明细表）
4
-- 租户过滤：仅统计 tenant-a 的数据
5
-- 聚合粒度：按门店（store_id）分组
6
-- 输出排序：按转化率（conversion_rate_pct）降序，即转化表现好的门店排在前面
7
-- ============================================================
8

9
SELECT
10
    store_id,
11
    -- 进入门店的独立会员数（去重 member_id）
12
    -- 事件条件：event_name = 'entry_detected'（检测到进入）
13
    uniqExactIf(member_id, event_name = 'entry_detected') AS entry_uv,
14
    -- 支付成功的订单数（去重 order_id）
15
    -- 事件条件：event_name = 'payment_success'（支付成功）
16
    uniqExactIf(order_id, event_name = 'payment_success') AS payment_orders,
17
    -- 支付失败的事件次数（非去重，每次失败计一条）
18
    -- 事件条件：event_name = 'payment_failed'（支付失败）
19
    countIf(event_name = 'payment_failed') AS payment_failed,
20
    -- 计算转化率（支付订单数 / 进入UV * 100），保留两位小数
21
    -- 使用 nullIf(entry_uv, 0) 防止除零错误，当 entry_uv 为 0 时返回 NULL，使结果为 NULL
22
    round(payment_orders / nullIf(entry_uv, 0) * 100, 2) AS conversion_rate_pct
23
FROM analytics.store_events
24
WHERE
25
    tenant_id = 'tenant-a'
26
    AND event_date = '2026-07-21'
27
GROUP BY store_id          -- 按门店聚合，每个门店输出一行汇总结果
28
ORDER BY conversion_rate_pct DESC;   -- 按转化率从高到低排序

1
-- ============================================================
2
-- 查询指定日期（2026-07-21）各门店各设备上支付失败事件的详细分布
3
-- 目的：定位高频失败组合（门店 + 设备 + 软件版本 + 错误码），辅助排查系统问题
4
-- 数据来源：analytics.store_events（门店事件明细表）
5
-- 租户过滤：仅统计 tenant-a 的数据
6
-- 聚合粒度：按门店、设备、软件版本、错误码 四个维度组合
7
-- 输出排序：先按失败次数降序（问题最严重的排前面），再按最大耗时降序（性能最差的排前面）
8
-- ============================================================
9

10
SELECT
11
    store_id,
12
    device_id,
13
    -- 从 properties 字段中解析出软件版本号（字符串类型）
14
    -- properties 为 JSON 格式，示例：{"software_version":"v2.3.1","error_code":"E001","cost_ms":150}
15
    JSONExtractString(properties, 'software_version') AS software_version,
16
    -- 从 properties 字段中解析出错误码（字符串类型）
17
    -- 用于标识具体的失败原因，如超时、签名错误、库存不足等
18
    JSONExtractString(properties, 'error_code') AS error_code,
19
    -- 统计该组合下支付失败事件的次数（count() 计数所有行，无需去重，因为每条事件即一次失败）
20
    count() AS failed_count,
21
    -- 从 properties 字段中解析出耗时（毫秒，无符号整数），并取该组合下的最大值
22
    -- 用于观察最慢的那次失败请求耗时，辅助判断是否因性能问题导致超时失败
23
    max(JSONExtractUInt(properties, 'cost_ms')) AS max_cost_ms
24
FROM analytics.store_events
25
WHERE
26
    tenant_id = 'tenant-a'
27
    AND event_date = '2026-07-21'
28
    AND event_name = 'payment_failed'
29
GROUP BY
30
    store_id,               -- 按门店分组，看问题是否集中在某个门店
31
    device_id,              -- 按设备分组，看是否某台设备硬件故障或网络差
32
    software_version,       -- 按软件版本分组，看是否新版本引入的缺陷
33
    error_code              -- 按错误码分组，看具体失败原因类型
34
ORDER BY
35
    failed_count DESC,      -- 失败次数最多的组合排在最前，优先关注高发问题
36
    max_cost_ms DESC;       -- 若失败次数相同，耗时最大的排前面，优先解决性能瓶颈

1
-- ============================================================
2
-- 查询特定门店（store-hz-002）在指定日期（2026-07-21）内，
3
-- 按小时统计各核心事件（进店、支付成功、支付失败）的发生趋势。
4
-- 数据来源：analytics.store_event_1h（按小时预聚合的中间表 / 物化视图）
5
-- 作用：用于观察一天中不同时段的流量和支付转化波动，辅助运营排班或系统压测分析
6
-- ============================================================
7

8
SELECT
9
    hour,
10
    event_name,
11

12
    -- 【重要】合并预聚合的事件计数
13
    -- 该表使用 AggregatingMergeTree 引擎，原数据已通过 countState() 预先聚合。
14
    -- countMerge() 用于将底层多个数据块中的中间状态合并为最终的实际计数值。
15
    countMerge(event_count) AS event_count,
16

17
    -- 【重要】合并预聚合的独立访客数（UV）
18
    -- 原数据已通过 uniqCombined64State() 预先进行了近似去重。
19
    -- uniqCombined64Merge() 用于将底层存储的中间状态合并为最终的近似 UV 值。
20
    -- 注意：这是近似去重算法（HyperLogLog 变种），性能极高，适合大数据量场景。
21
    uniqCombined64Merge(member_uv) AS member_uv
22

23
FROM analytics.store_event_1h
24

25
WHERE
26
    tenant_id = 'tenant-a'
27
    AND store_id = 'store-hz-002'
28
    AND event_name IN ('entry_detected', 'payment_success', 'payment_failed')
29
    AND hour >= '2026-07-21 00:00:00'
30
    AND hour < '2026-07-22 00:00:00'
31
GROUP BY hour, event_name
32
ORDER BY hour, event_name;