ClickHouse学习笔记九副本与分片

集群是副本和分片的基础。副本防止数据丢失（冗余存储），分片实现数据水平切分。

1.1 概述

ClickHouse 集群配置非常灵活：

• 单集群或多集群

• 每个集群的节点、分区和副本数量可以各不相同

• 副本 = 数据完全相同的多个节点

• 分片 = 数据完全不同的多个节点

1.2 数据副本

使用 ReplicatedMergeTree 系列表引擎实现副本能力。

1.2.1 副本的特点

1.2.2 ZooKeeper 的配置方式

在 /etc/clickhouse-server/config.d/metrika.xml 中配置：

<yandex>
    <zookeeper-servers>
        <node index="1">
            <host>hdp1.nauu.com</host>
            <port>2181</port>
        </node>
    </zookeeper-servers>
</yandex>

在 config.xml 中引入：

<include_from>/etc/clickhouse-server/config.d/metrika.xml</include_from>
<zookeeper incl="zookeeper-servers" optional="false"/>

通过 system.zookeeper 代理表查询 ZooKeeper：

SELECT * FROM system.zookeeper WHERE path = '/';
SELECT * FROM system.zookeeper WHERE path = '/clickhouse';

1.2.3 副本的定义形式

ENGINE = ReplicatedMergeTree('zk_path', 'replica_name')

命名约定：

/clickhouse/tables/{shard}/table_name

• /clickhouse/tables/：固定前缀

• {shard}：分片编号

• table_name：表名

replica_name： 每个副本唯一，通常使用服务器域名。

-- 1 分片 1 副本
ReplicatedMergeTree('/clickhouse/tables/01/test_1', 'ch5.nauu.com')
ReplicatedMergeTree('/clickhouse/tables/01/test_1', 'ch6.nauu.com')
​
-- 多分片 1 副本
ReplicatedMergeTree('/clickhouse/tables/01/test_1', 'ch5.nauu.com')
ReplicatedMergeTree('/clickhouse/tables/01/test_1', 'ch6.nauu.com')
ReplicatedMergeTree('/clickhouse/tables/02/test_1', 'ch7.nauu.com')
ReplicatedMergeTree('/clickhouse/tables/02/test_1', 'ch8.nauu.com')

1.3 ReplicatedMergeTree 原理解析

1.3.1 ZooKeeper 数据结构

/clickhouse/tables/{shard}/{table}/
├── metadata             -- 主键、分区键、采样表达式
├── columns              -- 列字段信息
├── replicas             -- 副本名称列表
├── leader_election      -- 主副本选举
├── blocks               -- Block 的 Hash 摘要（去重）
├── block_numbers        -- 按写入顺序记录
├── quorum               -- 写入确认数量（insert_quorum）
├── log                  -- 操作日志（持久顺序型）
│   ├── log-0000000000
│   └── log-0000000001
├── mutations            -- MUTATION 操作日志
│   └── 0000000000
└── replicas/{replica_name}/
    ├── queue            -- 任务队列
    ├── log_pointer      -- 日志指针
    └── mutation_pointer -- MUTATION 指针

LogEntry 核心属性

MutationEntry 核心属性

1.3.2 副本协同的核心流程

INSERT 流程

1. 创建副本实例： 初始化 ZK 节点 → 注册副本 → 参与选举

2. 创建其他副本实例： 同样注册和选举

3. 写入数据： 本地写入分区 → 记录 block_id → 去重检查

4. 推送 Log 日志： 执行 INSERT 的副本向 /log 推送（type=get）

5. 拉取 Log 日志： 其他副本监听 → 更新 log_pointer → 入队

6. 下载数据： 选择最优远端副本（log_pointer 最大 + queue 最少）

7. 响应下载： 通过 DataPartsExchange 服务发送分区数据

8. 本地写入： 写入临时目录 → 重命名完成

原则： ZooKeeper 不传输表数据，只分发指令。数据通过 HTTP 点对点传输。

MERGE 流程

1. 非主副本远程连接到主副本

2. 主副本接收连接

3. 主副本制定合并计划 → 推送 Log（type=merge）

4. 各副本拉取 Log 到队列

5. 各副本在本地执行 MERGE

合并由主副本统一制定计划，各副本本地执行。

MUTATION 流程

1. 执行 ALTER DELETE/UPDATE → 创建 MutationEntry 到 ZK /mutations

2. 所有副本监听 mutations 节点

3. 只有主副本响应 → 转换为 LogEntry 推送到 /log（type=mutate）

4. 各副本拉取 Log 到队列

5. 各副本在本地执行 MUTATION

ALTER 流程

1. 修改 ZK 共享元数据（/metadata, /columns）→ 版本号提升

2. 各副本监听变更 → 本地执行更新

3. 确认所有副本完成

ALTER 不涉及 /log 分发，直接修改共享元数据。

1.4 数据分片

数据分片需要结合 Distributed 表引擎使用。

1.4.1 集群的配置方式

形式一：不包含副本的分片（使用 node 标签）

<clickhouse_remote_servers>
    <shard_2>
        <node>
            <host>ch5.nauu.com</host>
            <port>9000</port>
        </node>
        <node>
            <host>ch6.nauu.com</host>
            <port>9000</port>
        </node>
    </shard_2>
</clickhouse_remote_servers>

形式二：自定义分片与副本（使用 shard/replica 标签）

<!-- 2 分片 0 副本 -->
<sharding_simple>
    <shard>
        <replica><host>ch5.nauu.com</host><port>9000</port></replica>
    </shard>
    <shard>
        <replica><host>ch6.nauu.com</host><port>9000</port></replica>
    </shard>
</sharding_simple>
​
<!-- 1 分片 1 副本 -->
<sharding_simple_1>
    <shard>
        <replica><host>ch5.nauu.com</host><port>9000</port></replica>
        <replica><host>ch6.nauu.com</host><port>9000</port></replica>
    </shard>
</sharding_simple_1>
​
<!-- 2 分片 1 副本（高可用） -->
<sharding_ha>
    <shard>
        <replica><host>ch5.nauu.com</host><port>9000</port></replica>
        <replica><host>ch6.nauu.com</host><port>9000</port></replica>
    </shard>
    <shard>
        <replica><host>ch7.nauu.com</host><port>9000</port></replica>
        <replica><host>ch8.nauu.com</host><port>9000</port></replica>
    </shard>
</sharding_ha>

验证集群配置：

SELECT cluster, host_name FROM system.clusters;

1.4.2 基于集群实现分布式 DDL

CREATE TABLE test_1_local ON CLUSTER shard_2 (
    id UInt64
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/test_1', '{replica}')
ORDER BY id;
​
DROP TABLE test_1_local ON CLUSTER shard_2;

宏变量： 在 config.xml 中为每个节点定义：

<macros>
    <shard>01</shard>
    <replica>ch5.nauu.com</replica>
</macros>

分布式 DDL 在 ZK 的路径：/clickhouse/task_queue/ddl

DDLLogEntry 核心属性

1.5 Distributed 原理解析

1.5.1 定义形式

ENGINE = Distributed(cluster, database, table [, sharding_key])

完整示例：

-- 分布式表
CREATE TABLE test_shard_2_all ON CLUSTER sharding_simple (
    id UInt64
) ENGINE = Distributed(sharding_simple, default, test_shard_2_local, rand());
​
-- 本地表
CREATE TABLE test_shard_2_local ON CLUSTER sharding_simple (
    id UInt64
) ENGINE = MergeTree()
ORDER BY id
PARTITION BY id;

1.5.2 查询分类

1.5.3 分片规则

分片键计算：

slot = shard_value % sum_weight

• shard_value：分片键取值

• sum_weight：所有分片权重之和

• slot 按权重区间映射到对应分片

-- 按用户 ID 的余数
Distributed(cluster, database, table, userid)
-- 按随机数
Distributed(cluster, database, table, rand())
-- 按散列值
Distributed(cluster, database, table, intHash64(userid))

1.5.4 分布式写入流程

1. 接收 INSERT → 按分片规则划分数据

2. 本地分片数据直接写入本地表

3. 远端分片数据写入临时 bin 文件（/database@host:port/[num].bin）

4. DirectoryMonitor 线程异步发送远端数据（传输前压缩）

5. 远端节点接收并写入本地表

6. 确认完成

同步/异步： insert_distributed_sync 参数控制（默认 false=异步）

副本数据复制的两种方式

<shard>
    <internal_replication>true</internal_replication>
    <replica>...</replica>
    <replica>...</replica>
</shard>

1.5.5 分布式查询流程

多副本路由规则（load_balancing）

多分片查询

1. 接收 SELECT → 转换为各分片的子查询

2. 并行查询本地（One）和远端分片（Remote）

3. 汇总各分片结果（Aggregator）

GLOBAL 优化分布式子查询

问题：不使用 GLOBAL 时查询放大 N² 倍

-- ❌ 查询放大 N² 倍（N = 分片数量）
SELECT uniq(id) FROM test_query_all WHERE repo = 100
AND id IN (SELECT id FROM test_query_all WHERE repo = 200);
​
-- ✅ 使用 GLOBAL 优化
SELECT uniq(id) FROM test_query_all WHERE repo = 100
AND id GLOBAL IN (SELECT id FROM test_query_all WHERE repo = 200);

GLOBAL IN 流程：

1. 将 IN 子句单独提出，发起一次分布式查询

2. 汇总子句结果 → 放入临时 Memory 表

3. 将 Memory 表分发到远端分片节点

4. 各分片使用本地临时数据执行完整 SQL

5. 汇总最终结果

⚠️ GLOBAL 在网络间分发数据，IN/JOIN 子句返回数据不宜过大。可使用 DISTINCT 去重。

1.6 小结