概述
不同于replicated模式，distributed有shard的概念，即一张表的完整数据并不存放在一个物理节点上，而是分布在多个不同的物理节点。Distributed引擎本身不存储数据，不过它支持在多台server上进行分布式的，并行的查询。比如一张distributed表有3个shard，分布在3个不同的server上面，当查询请求发到其中一台server（server1）的时候，该server会同时向其他两台server（server2和server3）发送数据请求，另外两台server处理完数据以后会将结果返回server1，在server1再进行处理后将最终结果返回客户端。
配置
clickhouse有几种写配置的方式，一种是直接在/etc/clickhouse-server/config.xml里面写，另一种是手动创建/etc/metrika.xml文件，这样原config.xml中带有incl属性的element会被相关联的value替换掉。还有一种配置方式是在element中加"from_zk="/path/to/node""属性，将xml配置存放在zookeeper中。zookeeper中的xml会成为config.xml中element的子内容。
下面采用的是在默认配置文件/etc/clickhouse-server/config.xml进行配置的方式。
1.同replicated模式，zookeeper是必须配置的
2.配置remote-servers列表，
<remote_servers>
<test_2shard_2replica>
<shard>
<weight>1</weight>
<internal_replication>true</internal_replication>
<replica>
<host>172.18.164.98</host>
<port>9000</port>
<user>default</user>
<password>compass</password>
</replica>
<replica>
<host>172.18.171.101</host>
<port>9000</port>
</replica>
</shard>
<shard>
<weight>1</weight>
<internal_replication>true</internal_replication>
<replica>
<host>172.18.171.101</host>
<port>9000</port>
</replica>
<replica>
<host>172.18.164.98</host>
<port>9000</port>
<user>default</user>
<password>compass</password>
</replica>
</shard>
</test_2shard_2replica>
</remote_servers>
标签说明：
-<test_2shard_2replica>：cluster名称，该标签的名字可以自定义，只需符合xml标签命名规则即可。该标签名字会在创建distributed引擎表的时被引用
-<shard>：有多少并列的<shard>标签，就意味着一张distributed表有多少分片
-<weight>1</weight>：分片权重标签，默认值为1，引擎会根据这个值分发不同的数据量到shard上。比如，有两个分片，其中一个分片<weight>设置为4，另外一个分片<weight>设置为5，则4/(4+5) = 4/9的数据会被分发到第一个分片，另一个分片会被分配5/9的数据
-<internal_replication>：当local表使用replicated引擎时，该标签的值设置为true，往distributed表中写数时，会数据写在其中一个健康的replica中，然后各个replica之间通过zookeeper自动同步数据。其余情况设置成false，数据会往所有replica中写，这种情况下replica之间的数据没有进行一致性校验，假以时日不同replica之间的数据可能会有微小的差异。默认被设置为false
-<replica>：副本标签，其中<host>表示server的地址，<port>是tcp通信端口，一般是9000，如果目标服务器设置了用户名和密码，还需要配置<user>标签和<password>标签，通过system.clusters表可以查看server上的分布式配置信息
定义distributed表
1.首先在不同的分片server上创建本地表，示例：
CREATE TABLE IF NOT EXISTS default.customer_shop_local (shop_id UInt64, pin String, score Float64, sex String, age String, marital_status String, region String, pay_mode String, purchase_power String, commet_sensitive String, user_level String, promotion_sensitive String, type UInt8, date Date) ENGINE = MergeTree(date, (shop_id, pin, type), 2048);
2.然后在根据需求在server上创建distributed表（比如有3台分片server，只在一台server上创建了distributed表，则只有这台server可以提供该distributed表的服务，如果所有server都创建了distributed表，那么3台都可以提供服务）。
Distributed引擎创建template：Distributed(cluster, datebase, local_table[, sharding_key])，
其中：
-cluster需要写成在config里自定义的cluster名称
-database是分片数据库的名称
-local_table是分片本地表的名称
-最后一项sharding_key是选填的，可以是一个表达式，例如rand()，也可以是某列 如user_id，不过该列必须是integer类型，通过对该具体的值进行取余进行分片，如果担心这样没法均匀的进行分片，也可以加上hash函数，如intHash64(user_id)
示例：
CREATE TABLE IF NOT EXISTS customer_shop_all AS customer_shop_local ENGINE = Distributed(test_2shard_2replica, default, customer_shop_local, rand())
