副本的目的防止数据丢失，保证高可用，分片则是实现数据的水平切分。
使用副本需要使用replicatedMergeTree存储引擎。MergeTree存储引擎存储数据时首先将数据写入内存缓冲区，然后数据被写入本地磁盘临时目录分区，待全部完成后再将临时目录重新命名为正式分区。
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')
使用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}:每个副本各自的节点下的一组监听节点，用于指导副本在本地执行具体的任务指令，其中重要的节点为：
log、mutation记录变化，replica负责操作任务的本地执行
/log和/mutation是分发操作指令的信息通道，发送指令的方式，则是为这些父节点添加子节点。所有的副本实例，都会监听主副本父节点的变化，当有子节点被添加，他们能实时感知。
以insert为例：
对于insert，当在master执行插入语句时，master负责向/log节点推送操作日志，如日志为/log/log-0000000；
其他副本会一直监听/log节点变化，当master推送了/log/log-00000000之后，node1会触发日志的拉取任务更更新log_pointer,将其指向最新日志下标；
node1拉取之后，并不会直接执行，而是将任务对象放入队列；
node1基于/queue执行任务，node1从/replicas节点拿到所有的副本节点，选择一个log_pointer值最大(大意味着该副本执行的日志最多，数据更完整)，/queue最小(小意味着该副本当前的任务执行负担小)作为自己数据的副本来源
向该副本请求下载数据
默认情况下，create、drop、rename、alter等不支持分布式执行，分布式DDL在zookeeper内使用的路径为/clickhouse/task_queue/ddl。该路径由users.xml中的distributed_dll指定
/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
假设建表语句在master上执行，步骤如下：
1）master负责创建DDLLogEntry日志并将日志推送到zookeeper,同时由这个节点负责监控任务的执行进度。假设日志为/ddl/query-0000000002
2）masrer,node1,node2监控/ddl/query-0000000002日志的推送，然后拉取日志到本地。首先判断本机host是否在DDLLogEntry的hosts列表中。若在，则执行，执行完毕后，将状态写入finished节点；不包含，则推送。
3）第一步推送日志之后，客户端会阻塞180秒，期望所有的节点执行完毕。若等待时间大于180秒，会转入后台线程继续等待。等待时间
分布式表与本地表之间的表结构在创建时不会进行检查，只有在查询时才会进行检查。
引擎形式为：
engine=Distributed(cluster,database,table,sharding_key);公式为：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节点的分布式表负责切分数据，并向所有分片节点发生数据。
有2种方式：由distributied引擎负责将数据写入副本；借助replicatedMergeTree表引擎实现副本数据