写入链路与吞吐调优：从单条 INSERT 到批量装载

写入链路：
□ 客户端发 INSERT → 按列组装内存 Block
□ 攒够 min_insert_block_size_rows（约 100 万行）
□ Block 压缩编码 → 写列文件 + 索引文件
□ 落盘为不可变 Part，后台异步 merge

为什么单条 INSERT 慢：
□ 每条 1 行 → Block 永远攒不满
□ 每 INSERT 建 Part、写索引、fsync，海量小 part
结论：写入单元 = 批量 Block，不是单行

-- 观察最近写入操作
SELECT event_time, query, written_rows, query_duration_ms
FROM system.query_log
WHERE query_kind = 'Insert'
ORDER BY event_time DESC LIMIT 20;

-- 当前 parts 总数：小 part 多的信号
SELECT count() AS parts, sum(rows) AS total_rows
FROM system.parts WHERE active = 1 AND table = 'events';

批量的收益：
□ 单次 Block 越大 → 压缩率越高
□ 写入次数少 → 建 part 开销分摊
□ 网络往返少 → RTT 不再是瓶颈
□ Part 数量少 → 后续 merge 压力小

经验值：
□ 单批 10 万~100 万行（或 100MB~1GB）
□ 用一条 SQL 带多行 VALUES
□ 避免逐行 insertOne
□ 批次太大内存上升 → 按行数/字节拆批

-- 正例：一条 SQL 批量插入
INSERT INTO events (event_time, user_id, event_type, value)
VALUES
    ('2026-09-30 10:00:00', 1, 'view', 10),
    ('2026-09-30 10:00:01', 2, 'click', 20),
    ...;  -- 一次带十万行

-- 更优：从文件批量导入
INSERT INTO events SELECT * FROM file('events.csv', 'CSVWithNames');

async_insert 原理：
□ 开启后 INSERT 先进内存缓冲
□ 服务器按超时/大小把缓冲合并成块落盘
□ 对外表现为「每条成功」，实则批量落盘

关键设置：
□ async_insert = 1：开启异步缓冲
□ wait_for_async_insert = 1：等待真正落盘
□ async_insert_max_data_size：缓冲上限（1MB）
□ async_insert_busy_timeout_ms：攒批时间窗（1s）

权衡：吞吐高 RTT 低；wait=0 崩溃时缓冲数据可能丢失

SET async_insert = 1;
SET wait_for_async_insert = 1;

-- 多条小 INSERT 会被服务器攒批落盘
INSERT INTO events (event_time, user_id, event_type, value) VALUES ('2026-09-30 10:00:00', 1, 'view', 10);
INSERT INTO events (event_time, user_id, event_type, value) VALUES ('2026-09-30 10:00:00', 1, 'click', 20);

-- 观察异步缓冲产生的落盘效果
SELECT query, written_rows, query_duration_ms, memory_usage
FROM system.query_log
WHERE query_kind = 'Insert' AND event_time > now() - INTERVAL 10 MINUTE;

诊断指标：
□ written_rows / written_bytes：单次写入量
□ query_duration_ms：写入耗时
□ memory_usage：写入内存峰值

看什么：
□ 单批行数小 → 批太小，合并次数多
□ 耗时与行数不成比例 → 网络/磁盘 fsync
□ parts 增长快 → merge 追不上

工具：system.processes / system.merges / system.part_log

-- 最近 1 小时每次 INSERT 的画像
SELECT query_duration_ms, written_rows, written_bytes,
       ProfileEvents['InsertedRows'] AS inserted, memory_usage
FROM system.query_log
WHERE query_kind = 'Insert' AND event_time > now() - INTERVAL 1 HOUR
ORDER BY query_duration_ms DESC LIMIT 20;

-- 看是否有 merge 长期追不上
SELECT table, is_mutation, elapsed, rows_written
FROM system.merges ORDER BY elapsed DESC LIMIT 10;

parts 膨胀的传导链：
□ 每批数据进入所属分区 → 产生一个 part
□ 分区多 → part 基数大 → merge 排队
□ part 多 → 查询要读更多文件头
□ 极端：分区数 > 查询线程数 → 并行浪费

分区过多场景：按小时/天分区、每分区只写几十行

控制手段：
□ 分区粒度对齐查询与 TTL 粒度（天/月）
□ 控制单分区 part 数
□ OPTIMIZE FINAL 主动合并不该频繁做

-- 看各分区 part 数与行数分布
SELECT partition, count() AS parts, sum(rows) AS rows,
       sum(bytes_on_disk) AS bytes
FROM system.parts
WHERE active = 1 AND table = 'events'
GROUP BY partition ORDER BY parts DESC LIMIT 20;

-- 主动合并（应急手段，非日常）
OPTIMIZE TABLE events FINAL;

去重方案：
□ ReplicatedMergeTree 自带 block 去重
  → 相同 part 名只落盘一次
□ ReplacingMergeTree：按排序键去重
  → 相同 ORDER BY 键只留最新（按版本/时间）
  → 合并时生效，查询用 FINAL

适用边界：
□ block 去重防「同一批数据重复插入」
□ Replacing 去重防「同 key 多次更新累积」

注意：去重靠排序键/版本，不是任意列

-- ReplacingMergeTree：以 version 取最新
CREATE TABLE events_dedup (
    event_time DateTime, user_id UInt64,
    event_type String, version UInt32
) ENGINE = ReplacingMergeTree(version)
PARTITION BY toYYYYMM(event_time)
ORDER BY (user_id, event_type);

-- 重复写入同一 key：合并后只留 version 最大者
INSERT INTO events_dedup VALUES ('2026-09-30 10:00:00', 1, 'view', 1);
INSERT INTO events_dedup VALUES ('2026-09-30 10:00:00', 1, 'view', 2);

-- 强制看最新态
SELECT * FROM events_dedup FINAL;

Kafka 写入优化要点：
□ 增大单批拉取：每批攒几万~几十万行再写
□ 用 INSERT ... SELECT 从 kafka 引擎表装载
□ 并行消费：分区数 = 消费并行度

Materialized View 管道：
□ Kafka 引擎表 + 物化视图 → 目标 MergeTree
□ 视图「消费即写」，频率由 kafka_max_block_size 控制

注意：消费速度 > 写入速度 → 堆积，看 system.kafka_consumers lag

-- Kafka 引擎表：原始层
CREATE TABLE kafka_events (
    event_time DateTime, 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 = 4;

-- 物化视图：批量转存到 MergeTree
CREATE MATERIALIZED VIEW mv_events TO events AS
SELECT * FROM kafka_events;

-- 检查消费进度与堆积
SELECT topic, consumer, partition, last_pulled_offset
FROM system.kafka_consumers WHERE topic = 'events';

并发与分区的相互作用：
□ 并发 INSERT 到同一分区 → 更多 part
□ 均匀分区键 → 写入分散，merge 也分散
□ 极端热分区（当天日期）→ 热点写入

策略：
□ 时间分区天然「写最新分区」→ 热点不可避免
□ 避免高基数随机键分区 → 每批都进新分区
□ 大批装载用 max_insert_threads 并行
□ 写入前按分区键排序 → part 更整

-- 控制并行写入线程
SET max_insert_threads = 4;

-- 大批装载：按分区键排序后再并行写（减少 part 碎片）
-- 例如把输入按 event_time 排序，再分段 INSERT
SELECT partition, count() AS parts, sum(rows) AS rows
FROM system.part_log
WHERE table = 'events' AND event_time > now() - INTERVAL 1 DAY
GROUP BY partition;

测试方法论：
□ 固定数据量与机器，一次只改一个变量
□ 指标：行/秒、写入耗时；辅指标：parts 数、merge 耗时

测试矩阵：
□ 基准：逐行 INSERT（最差基线）
□ 批量 1k / 10k / 100k / 1M 行每批
□ async_insert 开关、分区粒度（天 vs 月）、并行线程数

判定标准：吞吐提升同时 parts 数不爆炸

-- 生成测试数据并批量装载
INSERT INTO events
SELECT now() - rand() % 86400, rand() % 1000000,
       ['view', 'click', 'buy'][rand() % 3 + 1], rand() % 1000
FROM numbers(100000000);

-- 用 query_log 量化本次装载（行/秒）
SELECT count() AS insert_count, sum(written_rows) AS total_rows,
       sum(written_rows) / (sum(query_duration_ms) / 1000) AS rows_per_s
FROM system.query_log
WHERE query_kind = 'Insert' AND query ILIKE '%numbers%'
  AND event_time > now() - INTERVAL 1 HOUR;

-- 装载后 parts 健康度
SELECT count() AS parts, sum(rows) AS rows, max(rows) AS max_part_rows
FROM system.parts WHERE active = 1 AND table = 'events';

写入优化清单：
□ 批次：单批 10 万~100 万行，一条 SQL
□ 异步：逐条场景开 async_insert，wait=1
□ 分区：粒度对齐查询与 TTL，避免过细
□ 并发：max_insert_threads + 按分区键排序
□ 幂等：Replicated 去重 + Replacing 版本列
□ Kafka：大 batch 消费 + 物化视图转存
□ 监控：query_log 画像 + system.merges + lag

一句记忆：写入快 = 次数少、批次大、part 整

-- 写入优化的最小体检包
SELECT query_duration_ms, written_rows, memory_usage
FROM system.query_log WHERE query_kind = 'Insert'
ORDER BY event_time DESC LIMIT 20;

SELECT count(), sum(rows) FROM system.parts WHERE active = 1 AND table = 'events';