1. 分布式表

在上一篇《ClickHouse实时分析（六）- ClickHouse数据副本详解》中，讲了副本保障数据高可用的作用以及配置方式。这一篇开始讲讲 ClickHouse 的分片，也就是分布式表。
副本虽然能够提高数据的可用性，降低丢失数据的风险，但是每台服务器实际上必须容纳全量数据，对数据的横向扩容没有解决。
要解决数据水平切分的问题，需要引入分片的概念。通过分片把一份完整的数据进行切分，不同的分片分布到不同的节点上，再通过Distributed表引擎把数据拼接起来一同使用。
Distributed表引擎本身不存储数据，有点类似于 MyCat 之于 MySql，成为一种中间件，通过分布式逻辑表来写入、分发、路由来操作多台节点不同分片的分布式数据。
注意：ClickHouse 的集群是表级别的，实际生产环境中，大部分做了高可用，但是通常不会做分片，避免降低查询性能以及操作集群的复杂性。

2. 集群写入读取流程

2.1 写入流程

2.2 读取流程

3. 分片配置

3.1 集群及副本规划

2 个分片，只有第一个分片有副本：

hadoop121	hadoop122	hadoop123
<macros> <shard>01</shard> <replica>rep_1_1</replica> </macros>	<macros> <shard>01</shard> <replica>rep_1_2</replica> </macros>	<macros> <shard>02</shard> <replica>rep_2_1</replica> </macros>

3.2 配置步骤并演示

在hadoop121的/etc/clickhouse-server/config.xml中指定<remote_servers>，将<remote_servers>标签内的内容全部注释，然后加入下面的内容：

<gmall_cluster> <!-- 集群名称-->
	<shard> <!--集群的第一个分片-->
	    <internal_replication>true</internal_replication>
	    <replica>	<!--该分片的第一个副本-->
	        <host>192.168.68.121</host>
	        <port>9000</port>
	    </replica>
	    <replica>	<!--该分片的第二个副本-->
	        <host>192.168.68.122</host>
	        <port>9000</port>
	    </replica>
	</shard>
	<shard> <!--集群的第二个分片-->
	    <internal_replication>true</internal_replication>
	    <replica>       <!--该分片的第一个副本-->
	        <host>192.168.68.123</host>
	        <port>9000</port>
	    </replica>
	</shard>
</gmall_cluster>

指定宏的配置：

<macros>
    <shard>01</shard> <!--不同机器放的分片数不一样-->
    <replica>rep_1_1</replica> <!--不同机器放的副本数不一样-->
</macros>

将hadoop121的/etc/clickhouse-server/config.xml同步到hadoop122和hadoop123上，并修改hadoop122和hadoop123宏的配置：
hadoop122
1
2
3
4
<macros>
<shard>01</shard>
<replica>rep_1_2</replica>
</macros>
hadoop123
1
2
3
4
<macros>
<shard>02</shard>
<replica>rep_2_1</replica>
</macros>
重启三台 ClickHouse 服务器，systemctl restart clickhouse-server（需要开启 Zookeeper，具体可参考《ClickHouse实时分析（六）- ClickHouse数据副本详解》）；

在hadoop121上执行建表语句：

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

在hadoop122和hadoop123上查看表是否创建成功：

在hadoop121上创建Distribute分布式表：

create table st_order_mt_all 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));

参数含义：
Distributed(集群名称，库名，本地表名，分片键)
分片键必须是整型数字，所以用 hiveHash 函数转换，也可以 rand()

在hadoop121上插入测试数据：

insert into st_order_mt_all 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');