引言：理解ClickHouse分片的核心价值
在现代大数据时代，ClickHouse作为一款高性能的列式分析型数据库，已经成为处理海量数据的首选工具。然而，随着数据量的爆炸式增长，单节点ClickHouse往往难以应对PB级别的数据存储和查询需求。数据分片（Sharding）作为分布式架构的核心技术，能够将数据水平切分到多个节点上，显著提升存储容量和查询性能。
数据倾斜和查询性能瓶颈是分布式系统中最常见的两大挑战。数据倾斜会导致某些节点负载过高，形成系统短板；而查询性能瓶颈则会直接影响用户体验和业务效率。本文将深入探讨ClickHouse数据分片的实战策略，帮助您构建稳定高效的分布式系统。
一、ClickHouse分片基础概念
1.1 什么是数据分片？
数据分片是将大型数据集按照特定规则拆分到多个物理节点的过程。在ClickHouse中，分片通过
Distributed
表引擎实现，它像一个虚拟的协调器，将查询路由到正确的分片节点。
— 创建分布式表示例
CREATE TABLE distributed_logs ON CLUSTER ‘{cluster}’
(
`timestamp` DateTime,
`user_id` UInt64,
`event_type` String,
`payload` String
)
ENGINE = Distributed(
‘cluster_name’, — 集群名称
‘logs’, — 本地表名
‘logs_local’, — 本地表在分片中的名称
cityHash64(user_id) — 分片键
);
1.2 分片与副本的区别
分片（Shard）
：数据水平切分，每个分片包含数据的一个子集，所有分片的数据总和构成完整的数据集。
副本（Replica）
：数据冗余备份，每个分片可以有多个副本，用于高可用和故障转移。
— 带副本的分片配置示例 (config.xml)
二、数据倾斜的识别与诊断
2.1 数据倾斜的典型表现
数据倾斜通常表现为：
某些节点CPU、内存、磁盘IO远高于其他节点
查询响应时间不稳定，时快时慢
数据写入速度不均匀
某些分片数据量远大于其他分片
2.2 诊断工具与方法
2.2.1 使用system.metrics监控
— 查看各节点的查询执行情况
SELECT
hostName(),
metric,
value
FROM system.metrics
WHERE metric IN (‘Query’, ‘InsertQuery’)
ORDER BY hostName();
— 查看磁盘使用情况
SELECT
name,
free_space,
total_space,
(free_space / total_space) * 100 AS free_percent
FROM system.disks;
2.2.2 分析分片数据分布
— 检查各分片数据行数分布
SELECT
shardNum(),
count() AS rows,
sum(bytes_on_disk) AS total_bytes,
avg(bytes_on_disk) AS avg_bytes
FROM system.parts
WHERE database = ‘default’ AND table = ‘your_table’
GROUP BY shardNum()
ORDER BY shardNum();
— 查看分片键的值分布
SELECT
cityHash64(user_id) % 10 AS shard_key,
count() AS cnt,
count() * 100.0 / sum(count()) OVER () AS percentage
FROM your_table
GROUP BY shard_key
ORDER BY cnt DESC;
2.3 数据倾斜的常见原因
分片键选择不当
：使用低基数字段或分布不均匀的字段
时间序列数据
：按时间分片导致新数据集中写入某个分片
业务热点
：某些用户或商品ID访问频率极高
随机性不足
：分片算法无法保证均匀分布
三、解决数据倾斜的实战策略
3.1 优化分片键选择
3.1.1 使用高基数字段
— ❌ 错误示例：使用低基数字段
ENGINE = Distributed(cluster, db, table, status) — status只有3个值
— ✅ 正确示例：使用高基数字段
ENGINE = Distributed(cluster, db, table, user_id) — user_id基数高
3.1.2 复合分片键
— 使用多个字段组合提高随机性
ENGINE = Distributed(
cluster,
db,
table,
cityHash64(user_id, order_id, timestamp)
)
3.1.3 哈希分片 vs 范围分片
— 哈希分片（推荐用于均匀分布）
ENGINE = Distributed(cluster, db, table, cityHash64(user_id))
— 范围分片（适用于时间序列查询优化）
— 需要配合自定义分片逻辑
3.2 应用层路由策略
3.2.1 手动路由写入
# Python示例：应用层控制写入路由
import clickhouse_connect
import random
def write_with_load_balancing(data, shard_count=4):
“””
应用层实现负载均衡，避免热点
“””
# 策略1：随机分片
shard_id = random.randint(0, shard_count – 1)
# 策略2：一致性哈希（推荐）
user_id = data[‘user_id’]
shard_id = hash(user_id) % shard_count
# 策略3：加权随机（针对已知热点）
weights = [0.1, 0.1, 0.4, 0.4] # 给热点分片更小权重
shard_id = random.choices(range(shard_count), weights=weights)[0]
# 连接到对应分片写入
client = clickhouse_connect.get_client(
host=f’shard-{shard_id}.clickhouse.local’,
port=8123
)
client.insert(‘logs’, [data])
3.2.2 动态调整分片权重
— 创建分片权重表
CREATE TABLE shard_weights (
shard_id UInt8,
weight Float32,
reason String
) ENGINE = Memory;
— 插入权重配置
INSERT INTO shard_weights VALUES
(0, 0.3, ‘热点用户分片’),
(1, 0.3, ‘热点用户分片’),
(2, 0.2, ‘普通分片’),
(3, 0.2, ‘普通分片’);
— 查询时根据权重路由
SELECT
*,
cityHash64(user_id) % 4 AS default_shard,
if(default_shard < 2, '热点', '普通') AS shard_type
FROM logs;
3.3 使用CollapsingMergeTree解决热点更新
-- 对于频繁更新的场景，使用CollapsingMergeTree
CREATE TABLE user_events_local ON CLUSTER '{cluster}' (
`user_id` UInt64,
`event_id` UInt64,
`timestamp` DateTime,
`sign` Int8 -- 1表示新增，-1表示删除
) ENGINE = CollapsingMergeTree(sign)
ORDER BY (user_id, event_id, timestamp);
-- 写入时先删除再新增，避免数据膨胀
INSERT INTO user_events_local VALUES (123, 1, now(), -1); -- 删除旧数据
INSERT INTO user_events_local VALUES (123, 1, now(), 1); -- 插入新数据
四、查询性能瓶颈分析
4.1 性能瓶颈的识别
4.1.1 使用query_log分析慢查询
-- 查找慢查询
SELECT
query_id,
query,
query_duration_ms,
read_rows,
result_rows,
memory_usage
FROM system.query_log
WHERE query_duration_ms > 1000
AND event_date = today()
ORDER BY query_duration_ms DESC
LIMIT 10;
— 分析查询执行计划
EXPLAIN ESTIMATE
SELECT count() FROM logs WHERE timestamp > now() – INTERVAL 1 HOUR;
4.1.2 查看查询执行详情
— 启用详细日志
SET send_logs_level = ‘trace’;
— 查看分片间数据传输
SELECT
query_id,
address,
port,
is_local,
shard_num
FROM system.clusters
WHERE cluster = ‘cluster_name’;
4.2 常见性能瓶颈场景
4.2.1 跨分片查询性能问题
— ❌ 低效：分布式表上直接聚合
SELECT user_id, count()
FROM distributed_table
WHERE timestamp > ‘2024-01-01’
GROUP BY user_id
ORDER BY count() DESC
LIMIT 10;
— ✅ 高效：先本地聚合再全局合并
SELECT user_id, sum(cnt) as total_cnt
FROM (
SELECT user_id, count() as cnt
FROM local_table
WHERE timestamp > ‘2024-01-01’
GROUP BY user_id
)
GROUP BY user_id
ORDER BY total_cnt DESC
LIMIT 10;
4.2.2 数据本地化不足
— ✅ 优化：使用GLOBAL IN减少网络传输
SELECT * FROM distributed_table
WHERE user_id GLOBAL IN (
SELECT user_id FROM local_table WHERE condition
);
— ✅ 优化：使用JOIN时确保小表在内存中
SELECT a.*, b.*
FROM distributed_table_a a
ALL LEFT JOIN local_table_b b ON a.user_id = b.user_id;
五、高级优化策略
5.1 物化视图预聚合
— 创建按小时预聚合的物化视图
CREATE MATERIALIZED VIEW mv_hourly_stats ON CLUSTER ‘{cluster}’
ENGINE = Distributed(cluster, db, hourly_stats, rand())
AS SELECT
toStartOfHour(timestamp) AS hour,
user_id,
count() AS event_count,
sumIf(value, type = ‘purchase’) AS total_purchase
FROM logs_local
GROUP BY hour, user_id;
— 查询时直接访问物化视图
SELECT
hour,
sum(event_count) AS total_events,
sum(total_purchase) AS total_revenue
FROM mv_hourly_stats
WHERE hour >= toStartOfHour(now() – INTERVAL 7 DAY)
GROUP BY hour
ORDER BY hour;
5.2 数据生命周期管理
— 创建TTL自动清理旧数据
CREATE TABLE logs_local ON CLUSTER ‘{cluster}’ (
`timestamp` DateTime,
`data` String
) ENGINE = MergeTree()
ORDER BY timestamp
TTL timestamp + INTERVAL 30 DAY
SETTINGS index_granularity = 8192;
— 按分区设置TTL（更灵活）
ALTER TABLE logs_local
MODIFY COLUMN timestamp DateTime TTL timestamp + INTERVAL 7 DAY;
5.3 索引优化
— 创建跳数索引（Skip Index）
ALTER TABLE logs_local
ADD INDEX idx_user_id user_id TYPE bloom_filter GRANULARITY 3;
— 创建最小-最大索引
ALTER TABLE logs_local
ADD INDEX idx_timestamp timestamp TYPE minmax GRANULARITY 1;
— 查看索引使用情况
SELECT
name,
type,
data_compressed_bytes,
data_uncompressed_bytes
FROM system.columns
WHERE table = ‘logs_local’;
六、监控与告警体系
6.1 关键指标监控
— 创建监控视图
CREATE VIEW IF NOT EXISTS shard_monitor AS
SELECT
hostName() AS node,
metric_name,
metric_value,
timestamp
FROM system.metrics
WHERE metric_name IN (
‘Query’,
‘InsertQuery’,
‘Merge’,
‘NetworkReceive’,
‘NetworkSend’
);
— 设置告警阈值
— CPU使用率 > 80%
— 内存使用率 > 85%
— 磁盘使用率 > 80%
— 查询延迟 P99 > 1000ms
6.2 分布式追踪
— 启用查询追踪
SET log_queries = 1;
SET log_query_threads = 1;
— 分析查询路径
SELECT
query_id,
event_time,
query_duration_ms,
read_rows,
result_rows,
memory_usage,
peak_memory_usage
FROM system.query_log
WHERE query_id = ‘your-query-id’
ORDER BY event_time;
七、实战案例：电商日志系统优化
7.1 场景描述
假设我们有一个电商平台的用户行为日志系统，每天产生500GB数据，存在以下问题：
用户ID分布不均，VIP用户产生大量日志
查询需要实时响应，但跨分片聚合慢
数据写入集中在白天高峰时段
7.2 优化方案
7.2.1 分片策略调整
— 原始方案（导致倾斜）
ENGINE = Distributed(cluster, logs, logs_local, user_id)
— 优化方案1：复合哈希
ENGINE = Distributed(cluster, logs, logs_local,
cityHash64(user_id, rand()) % 1000000
)
— 优化方案2：应用层预处理
— 在写入前对VIP用户ID进行特殊处理
def normalize_user_id(user_id):
if user_id in VIP_USERS:
# VIP用户ID映射到多个虚拟ID
return hash(f”{user_id}_{random.randint(0, 9)}”)
return user_id
7.2.2 查询优化
— 创建用户维度预聚合表
CREATE TABLE user_daily_stats_local ON CLUSTER ‘{cluster}’ (
`date` Date,
`user_id` UInt64,
`page_views` UInt32,
`clicks` UInt32,
`purchases` UInt32,
`revenue` Decimal(18,2)
) ENGINE = SummingMergeTree()
ORDER BY (date, user_id);
— 物化视图自动填充
CREATE MATERIALIZED VIEW mv_user_daily_stats ON CLUSTER ‘{cluster}’
TO user_daily_stats_local
AS SELECT
toDate(timestamp) AS date,
user_id,
countIf(event_type = ‘page_view’) AS page_views,
countIf(event_type = ‘click’) AS clicks,
countIf(event_type = ‘purchase’) AS purchases,
sumIf(amount, event_type = ‘purchase’) AS revenue
FROM logs_local
GROUP BY date, user_id;
— 查询时直接访问预聚合表
SELECT
date,
sum(page_views) AS total_views,
sum(purchases) AS total_orders,
sum(revenue) AS total_revenue
FROM user_daily_stats_local
WHERE date >= today() – INTERVAL 7 DAY
GROUP BY date
ORDER BY date;
7.2.3 写入优化
# 批量写入与异步合并
import asyncio
import clickhouse_connect
async def batch_insert(client, data_batch):
“””批量写入减少合并压力”””
# 按时间窗口批量写入
for hour in range(0, 24, 6): # 每6小时一批
hour_data = [d for d in data_batch if hour <= d['timestamp'].hour < hour + 6]
if hour_data:
client.insert('logs_local', hour_data)
await asyncio.sleep(0.1) # 避免瞬时压力
# 使用压缩传输
client = clickhouse_connect.get_client(
host='clickhouse.local',
compress=True # 启用压缩
)
八、最佳实践总结
8.1 分片键选择原则
高基数
：选择唯一值数量远大于分片数的字段
随机性
：确保数据分布均匀，避免热点
业务相关
：考虑查询模式，尽量让查询命中更少的分片
稳定性
：避免频繁变更分片键
8.2 查询优化清单
[ ] 尽量使用本地表而非分布式表进行聚合
[ ] 使用GLOBAL IN/JOIN减少网络传输
[ ] 利用物化视图进行预聚合
[ ] 合理设置索引和TTL
[ ] 控制返回数据量，使用LIMIT分页
8.3 监控告警配置
# Prometheus监控配置示例
scrape_configs:
- job_name: 'clickhouse'
static_configs:
- targets: ['node1:9363','node2:9363','node3:9363','node4:9363']
metrics_path: '/metrics'
# 告警规则
groups:
- name: clickhouse_alerts
rules:
- alert: ClickHouseHighCPU
expr: clickhouse_metrics_Query > 100
for: 5m
labels:
severity: warning
annotations:
summary: “ClickHouse查询数过高”
九、常见问题与解决方案
Q1: 如何处理分片后数据不均匀的问题？
A
: 使用
ALTER TABLE … MOVE PARTITION
手动迁移分区，或重建表并调整分片键。
Q2: 查询分布式表比单表慢很多怎么办？
A
:
检查是否所有分片都参与了查询
使用
EXPLAIN
分析执行计划
考虑将查询下推到本地表执行
增加
max_threads
参数
Q3: 如何在线扩容分片？
A
:
新增节点并配置集群
使用
ALTER TABLE … DETACH PARTITION
分离旧数据
重新分配数据到新分片
更新分布式表配置
结语
ClickHouse数据分片是一个系统工程，需要从分片键设计、写入策略、查询优化、监控告警等多个维度综合考虑。解决数据倾斜和查询性能瓶颈的关键在于：
预防优于治疗
：在设计阶段就选择合适的分片策略
数据驱动
：通过监控数据持续优化
分层架构
：利用物化视图、预聚合表分层处理
工具赋能
：善用ClickHouse提供的诊断工具
通过本文介绍的实战策略，相信您能够构建出高性能、高可用的ClickHouse分布式集群，有效应对数据倾斜和查询性能挑战。记住，没有银弹，最佳方案需要根据具体业务场景和数据特征来定制。