Clickhouse之ReplicatedMergeTree原理详解

Clickhouse之ReplicatedMergeTree原理详解

        ReplicatedMergeTree作为Clickhouse（下文简称CK）复制表系列的基础表引擎，包涵了CK数据副本最核心的逻辑，本文详细介绍了ReplicatedMergeTree的核心原理，以便于更好地去使用ReplicatedMergeTree系列的表引擎。

        在ReplicatedMergeTree的核心逻辑中，大量使用了ZooKeeper分布式协同的能力，来实现多个ReplicatedMergeTree副本实例之间的协同，包括主副本选举、副本状态感知、 *** 作日志分发、任务队列和BlockID去重判断等。在执行INSERT数据写入、MERGE分区合并、MUTATION数据修改、ALTER元数据修改的时候，都会涉及与ZooKeeper的通信。（注：在通信的过程中，不会涉及任何表数据的传输，在查询数据的时候也不会访问ZooKeeper）。

---

Zookeeper节点介绍

        ReplicatedMergeTree需要依靠ZooKeeper的事件监听机制来实现各个副本之间的协同，在每张ReplicatedMergeTree表的创建过程中，它会以zk_path为根路径，在ZooKeeper中为这张表创建一组监听节点：

1. /metadata：保存元数据信息，包括主键、分区键、采样表达式等。
2. /columns：保存列字段信息，包括列名称和数据类型。
3. /replicas：保存副本名称，对应server参数中的replica_name。
4. /leader_election：用于主副本的选举工作，主副本会主导MERGE和MUTATION *** 作（ALTER DELETe、ALTER UPDATE）。这些任务在主副本制定后通过ZooKeeper将消息事件分发至其他副本。
5. /blocks：记录Block数据块的Hash信息，以及对应的partition_id。通过Hash信息能够判断Block数据块是否重复；partition_id记录了需要同步的数据分区。
6. /block_numbers：按照分区的写入顺序记录partition_id。各副本在本地进行MERGE时，都会根据block_numbers进行执行。
7. /quorum：记录quorum的数量，当至少有quorum数量的副本写入成功后，整个写 *** 作才算成功。quorum的数量由insert_quorum参数控制。
8. /log：一些 *** 作的日志节点（INSERT、MERGE等），保存了副本需要执行的任务指令。log使用了ZooKeeper的持久顺序型节点，每条指令的名称以log-为前缀递增，例如log-0000000000、log-0000000001等。每一个副本实例都会监听/log节点，当有新的指令加入时，它们会将 *** 作指令加入各副本自己的任务队列，并执行任务。
9. /mutations：MUTATION *** 作日志节点，作用与log日志类似，当执行alert DELETE和alert UPDATE查询时， *** 作指令会被添加到这个节点下。mutations同样使用了ZooKeeper的持久顺序型节点，但是它的命名没有前缀，例如0000000000、0000000001等。
10. /replicas/{replica_name}/*：各副本自己节点下的一些监听节点，各副本在本地执行具体任务会使用到，比如：

        /queue：任务队列节点，当副本从/log或/mutations节点监听到 *** 作指令时，会将执行任务添加至该节点下，并基于队列顺序执行。

        /log_pointer：log日志节点的一个指针，用于记录当前最后一次执行的log日志的编号，例如log_pointer：4对应了/log/log-0000000003（从0开始计数）。

        /mutation_pointer：mutations日志节点的一个指针，用于记录最后一次执行的mutations日志编号，例如mutation_pointer：0000000000对应了/mutations/000000000。

---

副本协同流程

        下面会详细介绍执行INSERT数据写入、MERGE分区合并、MUTATION数据修改、ALTER元数据修改时的同步流程（通过CK日志结合znode节点信息进行分析）。

        首先，准备测试环境，使用ReplicatedMergeTree表引擎创建1分片、2副本的数据表，两台server的域名分别为：

bdap-clickhouse-001shard-01replica.cebbank.com（下文简称001shard-01replica）

bdap-clickhouse-001shard-02replica.cebbank.com（下文简称001shard-02replica）

在创建的过程中，ReplicatedMergeTree会进行一些初始化 *** 作：

1. 根据zk_path初始化所有的znode节点；
2. 在/replicas/节点下注册自己的副本实例；
3. 启动监听任务，监听/log日志节点；
4. 参与副本选举，选举出主副本，选举是通过向/leader_election下插入子节点，最先插入成功的副本就是主副本。

---

Insert的执行流程

        首先往001shard-01replica上的表写入数据，通过查看日志(Wrote block with ID '202105_xxx_xxx')，此时会向blocks节点写入该分区的block_id，如果后续写入相同的batch，会通过block_id进行去重，忽略该batch数据的写入，然后该副本会向/log节点写入 *** 作日志。查看/log/log-0000000000节点：

source replica: 001shard-01replica

block_id: 202105_xxx_xxx

type:get

partition_name:202105_0_0_0

        可以看到 *** 作类型为get，需要下载的分区是202105_0_0_0，其余所有副本都会基于Log日志以相同的顺序执行 *** 作，001shard-02replica监听到/log节点的变化，拉取日志并更新/log_pointer节点（/replicas/001shard-02replica/log_pointer:0），然后将该任务放入自己的队列下（/replicas/001shard-02replica/queue/）：

Pulling 1 entries to queue: log-0000000000 - log-0000000000

        当队列轮到该任务执行时，001shard-02replica向001shard-01replica发起HTTP请求，下载上面分区的数据：

Fetching part 202105_0_0_0 from replicas/001shard-01replica

Sending request to http://001shard-01replica:8123/?endpoint=DataPartsExchange

        001shard-01replica收到请求后，响应(Sending part 202105_0_0_0)，001shard-02replica收到数据后，写入到该副本的part。（注：insert由哪个副本发起的 *** 作，整体流程就会由该副本负责）。

---

Merge的执行流程

        当ReplicatedMergeTree触发分区合并时，就会进入该流程，无论merge *** 作是从哪个副本发起的，合并计划都会由主副本在负责执行。比如当前环境主副本是001shard-01replica，我们通过001shard-02replica（为了验证是主副本主导 *** 作）执行optimize *** 作，强制触发merge part，此时001shard-02replica会通过/replica节点找到当前分片对应的主副本001shard-01replica，并请求与其建立连接，主副本响应后建立连接：

Connection (001shard-01replica): Connecting. Database: default. User: default
Connected ClickHouse Follower replica version 21.1.0, revision:54443, database: default, user: default.

        然后由主副本判断哪些part需要合并，并且制定merge计划，将该计划推送到/log节点下(/log/log-0000000002)，各副本按照该计划进行合并：

Source replica: 001shard-01replica

type: merge 202105_0_0_0 202105_1_1_0 into 202105_0_1_1

        从节点信息中可以看到， *** 作类型为merge上述两个part的数据，同时主副本还会对日志 *** 作的情况进行监听：

Waiting for queue-0000000002 to disappear from 001shard-01replica queue

        此时两副本监听到/log节点下的变化，分别将日志拉取到各自本地，并且加入到各自的/queue队列中，然后两副本基于各自的/queue开始执行任务，在本地进行merge：

Pulling 1 entries to queue: log-0000000002 - log-0000000002

Executing log entry to merge parts 202105_0_0_0, 202105_1_1_0 to 202105_0_1_1

---

MUTATION的执行流程

        当执行ALTER DELETE或ALTER UPDATE *** 作的时候，就会进入MUTATION流程，与merge流程类似，无论mutation *** 作是从哪个副本发起的，都会先转到主副本进行 *** 作。当前主副本是001shard-01replica，为了验证，我从001shard-02replica发起mutation请求，在该副本上做ALTER DELETE的 *** 作，执行后，首先会创建MUTATION ID，并将MUTATION *** 作推送到/mutations/0000000000：

created mutation with ID 0000000000

source replica: 001shard-02replica

mutation_id:2

partition_id: 202105

commands: DELETE WHERe id = '1'

        此时两副本都监听到/mutations节点的变化，但只有主副本才会响应MUTATION *** 作。此时主副本会将mutation *** 作推送至/log节点(/log/log-0000000003)，可以看到 *** 作类型是mutate，将202105_0_1_1改成202105_0_1_1_2（最后的2指的是mutation_id）：

source replica: 001shard-01replica

type : mutate 202105_0_1_1 to 202105_0_1_1_2

        此时两副本都监听到/log下的变化，分别拉取log日志到本地，并加入到各自的/queue队列中，然后各副本在本地执行MUTATION：

Pulling 1 entries to queue: log-0000000003 - log-0000000003

Executing log entry to mutate part 202105_0_1_1 to 202105_0_1_1_2

Cloning part 202105_0_1_1 to tmp_clone_202105_0_1_1_2

Renaming temporary part tmp_clone_202105_0_1_1_2 to 202105_0_1_1_2

---

ALTER的执行流程

        当执行ALTER *** 作（如增删表字段等）会进入该流程，该流程的执行由提交ALTER *** 作的副本负责，ALTER的流程不涉及/log节点日志的分发，它会直接去Zookeeper上修改元数据节点（/metadta，/column）的信息，节点的版本号也会随之增加，同时还会去监听所有副本的完成情况：

Updated shared metadata nodes in ZooKeeper. Waiting for replicas to apply changes.

Version of metadata nodes in ZooKeeper changed. Waiting for structure write lock.

Waiting for 001shard-01replica to apply changes

Waiting for 001shard-02replica to apply changes

        此时两副本监分别听到元数据的变化，会与自己本地的版本号进行对比，当发现节点的版本号大于本地的版本号时，会在各自本地执行ALTER *** 作：

metadata changed in ZooKeeper. Applying changes locally.

Applied changes to the metadata of the table.