副本虽然能够提高数据的可用性，降低丢失风险，但是每台服务器实际上必须容纳全量数据，对数据的横向扩容没有解决。

要解决数据水平切分的问题，需要引入分片的概念。通过分片把一份完整的数据进行切分，不同的分片分布到不同的节点上，再通过 Distributed 表引擎把数据拼接起来一同使用。

Distributed 表引擎本身不存储数据，有点类似于 MyCat，成为一种中间件，通过分布式逻辑表来写入、分发、路由来操作多台节点不同分片的分布式数据。

注意：ClickHouse 的集群是表级别的，实际企业中，大部分做了高可用，但是没有用分片，避免降低查询性能以及操作集群的复杂性。

集群规划

以下是一个 3 分片 2 副本共 6 个节点

写入过程

s代表分片，r代表副本.internal_replication(内部复制)，为 false 就是非内部复制，hdp1 不止要写道 s1r1 还要给 d1r2 写一次；为 true 是内部复制，让 s1r1 和 s1r2 之间内部同步。该参数推荐为 true

读取过程

配置步骤

用三台机器来搭建一个集群，haddop103，haddop104，hadoop113，搭建得结构如下图

以上是一个 2分片，分片1有2个副本，分片2有1个副本。因为clickhouse本身的原因，没办法让副本互相拥有。

配置分片文件

外部文件配置，新建 shard.xml 在 /etc/clickhouse-server/config.d/ 下，注意配置分片的存储位置的host是不能重复的，也就意味着不能把分片相互写。以下先改的是 103 服务器。

<?xml version="1.0"?>
<yandex>
  <remote_servers>
    <!-- 集群名称-->
    <gmall_cluster> 
      <!--集群的第1个分片-->
      <shard> 
        <internal_replication>true</internal_replication>
        <!-- 分片1，副本1 -->
        <replica>
          <host>10.240.30.103</host>
          <port>9000</port>
        </replica>
        <!-- 分片1，副本2 -->
        <replica>
          <host>10.240.30.104</host>
          <port>9000</port>
        </replica>
      </shard>
      <!--集群的第2个分片-->
      <shard> 
        <internal_replication>true</internal_replication>
          <!-- 副本往113写一份-->
          <replica> 
            <host>10.240.30.113</host>
            <port>9000</port>
          </replica>
      </shard>
    </gmall_cluster>
  </remote_servers>
  <!-- zookeeper 集群 -->
  <zookeeper-servers>
    <node index="1">
      <host>10.240.30.113</host>
      <port>2181</port>
    </node>
  </zookeeper-servers>
  <!-- 分片和副本命名 -->
  <macros>
    <!-- 第1个分片 -->
    <shard>01</shard>
    <!-- 第1个分片，第1个副本 -->
    <replica>rep_1_1</replica>
  </macros>
</yandex>

修改所创建文件的权限，一般我们都是root操作的要改成 clickhouse

chown clickhouse:clickhouse shard.xml

到 /etc/clickhouse-server/config.xml 中增加

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

把配置分发到其他 ck 种，并修改配置文件 shard.xml 文件，以及权限配置等。

# 104
  <!-- 分片和副本命名 -->
  <macros>
    <!-- 第1个分片 -->
    <shard>01</shard>
    <!-- 第1个分片，第2个副本 -->
    <replica>rep_1_2</replica>
  </macros>
# 113
  <!-- 分片和副本命名 -->
  <macros>
    <!-- 第2个分片 -->
    <shard>02</shard>
    <!-- 第2个分片，第1个副本 -->
    <replica>rep_2_1</replica>
  </macros>

重新启动 ck 集群，然后用客户端连接进去，查看这个集群是否被加进去

show clusters;

在启动第三个集群的时候一直报错，但不影响使用。
2021.11.02 19:52:37.521614 [ 40814 ] {} <Error> ServerErrorHandler: Code: 101, e.displayText() = DB::NetException: Unexpected packet from client, Stack trace (when copying this message, always include the lines below):
如果谁遇到并解决了，希望在这里说下。

建表

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);

• 建表会自动同步到其他ck上
• on cluster 关键词固定的，后面跟配 shard.xml 置文件里面的集群名称
• ReplicatedMergeTree 副本引擎里面的 {shard} 和 {replica} 会从<macros>读取

DB::Exception: There are two exactly the same ClickHouse instances
意思是你的配置里有两个相同的实例，可以检查配置的 remote_servers 里分片是否有相同 host，ck不允许有相同host，因为不允许互相存有分片副本。

成功会显示如下信息，并可以在其他服务器上查询到所创建的表

┌─host──────────┬─port─┬─status─┬─error─┬─num_hosts_remaining─┬─num_hosts_active─┐
│ 10.240.30.113 │ 9000 │      0 │       │                   2 │                0 │
│ 10.240.30.103 │ 9000 │      0 │       │                   1 │                0 │
│ 10.240.30.104 │ 9000 │      0 │       │                   0 │                0 │
└───────────────┴──────┴────────┴───────┴─────────────────────┴──────────────────┘

删除集群表

删除集群表语法

drop table default.st_order_mt ON CLUSTER gmall_cluster;

在我们删除本地表和分布式表后，立即重建是没有问题的。唯一有问题的就是复制表，因为复制表需要在zookeeper上建立一个路径，存放相关数据。clickhouse默认的库引擎是原子数据库引擎，删除 Atomic 数据库中的表后，它不会立即删除，而是会在480秒后删除。由下面这个参数控制：

<database_atomic_delay_before_drop_table_sec>480</database_atomic_delay_before_drop_table_sec>

其他解决方案:

• 使用普通数据库而不是原子数据库。 create database … Engine=Ordinary.
• 使用uniq ZK路径。{uuid}/clickhouse/tables/{shard}-{uuid}/
• 减少 database_atomic_delay_before_drop_table_sec = 0

创建分布式表

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

• on cluster 关键词固定的，后面跟配 shard.xml 置文件里面的集群名称
• Distributed(集群名称,库名,本地表名,分片键)，分片键必须是整形数字，所以用 hiveHash 函数转换，也可以rand()

插入数据

insert into st_order_mt_all2 values
(201,'sku_001',1000.00,'2020-06-01 12:00:00') ,
(202,'sku_002',2000.00,'2020-06-01 12:00:00'),
(203,'sku_004',2500.00,'2020-06-01 12:00:00'),
(204,'sku_002',2000.00,'2020-06-01 12:00:00'),
(205,'sku_003',600.00,'2020-06-02 12:00:00');

分布式表不存入数据，插入的数据都是给本地表，也就是创建分布式表之前所创建的表插入数据。