clickhouse--副本、分片、Distributed

        副本的目的防止数据丢失，保证高可用，分片则是实现数据的水平切分。

       使用副本需要使用replicatedMergeTree存储引擎。MergeTree存储引擎存储数据时首先将数据写入内存缓冲区，然后数据被写入本地磁盘临时目录分区，待全部完成后再将临时目录重新命名为正式分区。

1、建表

     RelicatedMergeTree在MergeTree上增加了zookeeper,replicatedmergeTree会在zookeeper内创建一系列监听节点，并以此实现相互通信。zookeeper只负责通信，不需要传输数据。

     在集群上创建表

 create table st_order_mt on cluster gmall_cluster
(
id UInt32,
create table st_order_mt on cluster gmall_cluster (id UInt32,
sku_id String,
total_amount Decimal(16,2), create_time Datetime
) engine =ReplicatedMergeTree('/clickhouse/tables/{shard}/st_order_mt','{replica}')partition by toYYYYMMDD(create_time)primary key (id)order by (id,sku_id);

创建分布式表

create table st_order_mt_all2 on cluster clustersName
(
id UInt32,sku_id String,
total_amount Decimal(16,2), create_time Datetime
)engine = Distributed(clustersName,default, st_order_mt,hiveHash(sku_id));

ReplicatedMergeTree('zk_path','replica_name')

2、zookeeper内的节点结构

使用prettyzoo工具连接zookeeper

          创建表时，以zk_path为路径，在zookeeper中为这张表创建一组监听节点,以节点2为例，路径为/clickhouse/tables/02/st_order_mt。创建的节点如下：

                    

这些节点按类型可以分为：

1）元数据

metadata:保存元数据信息，包括主键、分区键、采样表达式等

columns:保存列字段信息，包括列名称和数据类型

2）判断标识

block_numbers:按照分区的写入顺序，以相同的顺序记录partition_id.各个副本在本地进行merge时，会依照相同的block_numbers顺序进行。

blocks:记录block数据块的hash信息摘要，以及对应的partition_id,通过hash摘要能够判断block数据库是否重复，通过partition_id能够找到需要同步的数据分区

leader_election:用于主副本的选举工作，主副本会主导merge和mutation操作,这些任务在主副本完成之后再借助zookeeper将消息分发至其他副本

quorum:记录quorum的数量，当至少有quorum数量的副本写入成功后 ，整个操作才算成功，quorum的数量由insert_quorum参数控制，默认为0

3）日志

/log:操作日志，是zookeeper工作机制中最重要的一环，保存了副本需要执行的任务指令。log使用zookeeper的持久顺序型节点，每条指令的名称以log-为前缀递增，如log-00000000,log-00000001等。每一个副本都会监听/log节点，当有新的指令加入时，他们会把指令加入各自的任务队列，并执行任务。

mutations:mutation操作日志节点，作用与log日志类似，当执行alter delete和alter update时，操作指令会被添加到这个节点。mutations同样使用了zookeeper的持久顺序型节点，但命名没有前缀，每条指令直接以递增数字的形式保存。

replicas/{replica_name}:每个副本各自的节点下的一组监听节点，用于指导副本在本地执行具体的任务指令，其中重要的节点为：

• queue:任务队列节点，用于执行具体的操作任务。当副本从log或mutations节点监听到操作指令时，会将任务添加到该节点下，并基于队列执行
• log_pointer:log日志节点指针，记录了最后一次执行的log日志下标信息。如log_pointer:4对应了log/log-00000003（从0开始计数）
• mutation_pointer:mutation日志指针节点，记录了最后一次执行的mutations日志名称，如replicas/mutation_pointer:0000000对应mutation/00000000

log、mutation记录变化，replica负责操作任务的本地执行

/log和/mutation是分发操作指令的信息通道，发送指令的方式，则是为这些父节点添加子节点。所有的副本实例，都会监听主副本父节点的变化，当有子节点被添加，他们能实时感知。

3、副本协作的流程

    以insert为例：

    对于insert，当在master执行插入语句时，master负责向/log节点推送操作日志，如日志为/log/log-0000000；

   其他副本会一直监听/log节点变化，当master推送了/log/log-00000000之后，node1会触发日志的拉取任务更更新log_pointer,将其指向最新日志下标；

   node1拉取之后，并不会直接执行，而是将任务对象放入队列；

   node1基于/queue执行任务，node1从/replicas节点拿到所有的副本节点，选择一个log_pointer值最大(大意味着该副本执行的日志最多，数据更完整)，/queue最小(小意味着该副本当前的任务执行负担小)作为自己数据的副本来源

   向该副本请求下载数据

4、分布式DDL

      默认情况下，create、drop、rename、alter等不支持分布式执行，分布式DDL在zookeeper内使用的路径为/clickhouse/task_queue/ddl。该路径由users.xml中的distributed_dll指定

 DDLlogEntry日志对象

      /query-[seq]下记录的日志信息由DDLLogEntry记录，如query-0000000001的信息如下：

version: 1
query: CREATE TABLE default.st_order_mt_all2 UUID 
\'95c4d7b5-d397-4eb1-95c4-d7b5d3971eb1\' 
ON CLUSTER clustersName 
(
`id` UInt32,`sku_id` String, 
`total_amount` Decimal(16, 2), 
`create_time` Datetime
)ENGINE = Distributed(clustersName, default, st_order_mt, hiveHash(sku_id))
hosts: ['centos%2Dlinux%2Dmaster%2Eshared:9000',
'centos%2Dlinux%2Dnode1%2Eshared:9000',
'centos%2Dlinux%2Dnode2%2Eshared:9000']
initiator: centos%2Dlinux%2Dmaster%2Eshared:9000

• query:记录了具体的执行语句
• hosts:记录了指定集群的hosts主机列表
• initiator:记录初始化host主机的名称，hosts主机列表的取值来自于初始化host节点上的集群。如此处为master节点

分布式DDL执行的核心流程

    假设建表语句在master上执行，步骤如下：

1）master负责创建DDLLogEntry日志并将日志推送到zookeeper,同时由这个节点负责监控任务的执行进度。假设日志为/ddl/query-0000000002

2）masrer,node1,node2监控/ddl/query-0000000002日志的推送，然后拉取日志到本地。首先判断本机host是否在DDLLogEntry的hosts列表中。若在，则执行，执行完毕后，将状态写入finished节点；不包含，则推送。

3）第一步推送日志之后，客户端会阻塞180秒，期望所有的节点执行完毕。若等待时间大于180秒，会转入后台线程继续等待。等待时间

5、Distrubted

     分布式表与本地表之间的表结构在创建时不会进行检查，只有在查询时才会进行检查。

    引擎形式为：

engine=Distributed(cluster,database,table,sharding_key);

• cluster:集群名称
• databse:数据库名称
• table:本地表的名称
• sharding_key：分片键，选填参数。在数据写入的过程中，分布式表会依据分片键的规则，将数据分布到各个host节点的本地表。分片键必须是整型数字，可以使用hivehash函数进行转换，也可以是rand()

5、1数据写入分片的过程

      公式为：slot=shard_value%sum_weight

    集群配置时<weight>val</weight> ,sum_weight是所有分片的权重之和，shard_value是分片键的取值。如三个分片的权重是10，20，30，则slot在[0,10)区间，对第一个分片写入，[10,20)则对第二个分片写入。

     以节点master,node为例，在master节点上写入数据，流程如下：

1）根据分片规则划分数据，将属于当前分片的数据写入本地表

2）将属于node1的数据以分区为单位，写入分布式表的存储目录下临时bin文件下：如此处的路径为：

/st_order_mt_all2/default/@centos-linux-node1.shard:9000/1.bin

然后与node1建立连接

3）向node2发送数据，数据在传输之前被压缩

4）接收数据并写入本地表

传输过程中由master节点的分布式表负责切分数据，并向所有分片节点发生数据。

5、2数据写入副本的过程

   有2种方式：由distributied引擎负责将数据写入副本；借助replicatedMergeTree表引擎实现副本数据