clickhouse| 02 副本| 集群读写操作

1. 副本

副本的目的主要是保障数据的高可用性，即使一台ClickHouse节点宕机，那么也可以从其他服务器获得相同的数据。

1.1 副本写入流程

 client从clickhouse-a节点写数据，提交写入日志给zookeeper，clickhouse-b有一个监听器来监听zookeeper接收到日志之后从clickhouse-a中下载数据。

1.2 副本的配置步骤

①

• 启动zookeeper集群 （hadoop102、 hadoop103、 hadoop104 三台机器）
• 在hadoop102的/etc/clickhouse-server/config.d目录下创建一个名为metrika.xml的配置文件,内容如下：

<?xml version="1.0"?>
<yandex>
    <zookeeper-servers>
        <node index="1">
            <host>hadoop102</host>
            <port>2181</port>
        </node>
        <node index="2">
            <host>hadoop103</host>
            <port>2181</port>
        </node>
        <node index="3">
            <host>hadoop104</host>
            <port>2181</port>
        </node>
    </zookeeper-servers>
</yandex>

② 在 hadoop02的/etc/clickhouse-server/config.xml中增加

　　<include_from>/etc/clickhouse-server/config.d/metrika.xml</include_from>

③ 将metrika.xml和 config.xml 同步到hadoop103和hadoop104上

• scp /etc/clickhouse-server/config.d/metrika.xml root@hadoop103:/etc/clickhouse-server/config.d/metrika.xml
• scp /etc/clickhouse-server/config.d/metrika.xml root@hadoop104:/etc/clickhouse-server/config.d/metrika.xml
• scp /etc/clickhouse-server/config.xml root@hadoop103:/etc/clickhouse-server/config.xml
• scp /etc/clickhouse-server/config.xml root@hadoop104:/etc/clickhouse-server/config.xml

④ 分别在hadoop102、hadoop103 和hadoop104上启动ClickHouse服务

注意：因为修改了配置文件，如果以前启动了服务需要重启

[kris@hadoop102 clickhouse]$ sudo systemctl start clickhouse-server
[kris@hadoop102 clickhouse]$ ps -ef | grep clickhouse
clickho+  15120      1  2 16:55 ?        00:00:02 /usr/bin/clickhouse-server --config=/etc/clickhouse-server/config.xml --pid-file=/run/clickhouse-server/clickhouse-server.pid
kris   15213   4745  0 16:56 pts/2    00:00:00 grep --color=auto clickhouse

副本只能同步数据，不能同步表结构，所以我们需要在每台机器上自己手动建表

###在hadoop102上创建如下表
create table t_order_rep (
    id UInt32,
    sku_id String,
    total_amount Decimal(16,2),
    create_time  Datetime
 ) engine =ReplicatedMergeTree('/clickhouse/tables/01/t_order_rep','rep_102')
   partition by toYYYYMMDD(create_time)
   primary key (id)
   order by (id,sku_id);

##在hadoop103上创建如下表
create table t_order_rep (
    id UInt32,
    sku_id String,
    total_amount Decimal(16,2),
    create_time  Datetime
 ) engine =ReplicatedMergeTree('/clickhouse/tables/01/t_order_rep','rep_103')
   partition by toYYYYMMDD(create_time)
   primary key (id)
   order by (id,sku_id);

#在hadoop104上创建如下表
create table t_order_rep (
    id UInt32,
    sku_id String,
    total_amount Decimal(16,2),
    create_time  Datetime
 ) engine =ReplicatedMergeTree('/clickhouse/tables/01/t_order_rep','rep_104')
   partition by toYYYYMMDD(create_time)
   primary key (id)
   order by (id,sku_id);
   
   
参数解释
ReplicatedMergeTree 中，
　　第一个参数是分片的zk_path一般按照： /clickhouse/table/{shard}/{table_name} 的格式写，如果只有一个分片就写01即可。
　　第二个参数是副本名称，相同的分片副本名称不能相同。

在hadoop102上执行insert语句

insert into t_order_rep values
(101,'sku_001',1000.00,'2020-06-01 12:00:00'),
(102,'sku_002',2000.00,'2020-06-01 12:00:00'),
(103,'sku_004',2500.00,'2020-06-01 12:00:00'),
(104,'sku_002',2000.00,'2020-06-01 12:00:00'),
(105,'sku_003',600.00,'2020-06-02 12:00:00');

测试：

hadoop102 :) select * from t_order_rep;
hadoop103 :) select * from t_order_rep;
hadoop104 :) select * from t_order_rep;
都可以看到数据，查询出结果，说明配置成功

2. 分片集群

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

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

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

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

2.1 集群写入流程（3分片2副本共6个节点）

s1、 s2、 s3是三个分片，相当于是把表中的数据拆分成三份 做三个分片；为了保证高可用性，每一个分片都对应一个副本；

s1分片的两个副本r1、 r2...；distribute  hdp1相当于是一个代理，来接收请求。

客户端发送写请求给distribute，有一个internal_replication的配置，false or true。

• 如果设置为true，只需要把数据写入到第一个分片的第一个副本即可，数据的同步由第一个副本来做；
• 如果设置为false，这两个副本的数据都是由distribute来发送。

2.2 集群读取流程（3分片2副本共6个节点）

一份数据存储了2份，有一个副本，当读取数据时从哪里读取数据呢？哪个副本中读取呢。 每个节点会记录一个犯错数；

当客户端发送读命令时，它会优先选择errors_count小的副本，errors_count相同的有随机、顺序、随机（优先第一顺位）、host名称近似等四种选择方式。

2.3 集群配置（3分片2副本共6个节点）

配置的位置还是在之前的/etc/clickhouse-server/config.d/metrika.xml，内容如下

<yandex>
    <clickhouse_remote_servers>
        <gmall_cluster> <!-- 集群名称--> 
            <shard>  <!--集群的第一个分片-->
                <internal_replication>true</internal_replication>
                <!--该分片的第一个副本-->
                <replica>    
                    <host>hadoop201</host>
                    <port>9000</port>
                 </replica>
                 <!--该分片的第二个副本-->
                 <replica> 
                    <host>hadoop202</host>
                    <port>9000</port>
                 </replica>
            </shard>

              <shard>  <!--集群的第二个分片-->
                 <internal_replication>true</internal_replication>
                 <replica>    <!--该分片的第一个副本-->
                    <host>hadoop203</host>
                     <port>9000</port>
                 </replica>
                 <replica>    <!--该分片的第二个副本-->
                    <host>hadoop204</host>
                    <port>9000</port>
                 </replica>
              </shard>

              <shard>  <!--集群的第三个分片-->
                 <internal_replication>true</internal_replication>
                 <replica>     <!--该分片的第一个副本-->
                    <host>hadoop205</host>
                    <port>9000</port>
                 </replica>
                 <replica>    <!--该分片的第二个副本-->
                    <host>hadoop206</host>
                    <port>9000</port>
                 </replica>
              </shard>
        </gmall_cluster>
    </clickhouse_remote_servers>
</yandex>

2.4 配置3节点版本集群及副本

集群及副本规划（2个分片，只有第一个分片有副本）

[root@hadoop102 config.d]# vim /etc/clickhouse-server/config.d/metrika-shard.xml

<?xml version="1.0"?>
<yandex>
    <clickhouse_remote_servers>
        <gmall_cluster> <!-- 集群名称--> 
            <shard>         <!--集群的第一个分片-->
            <internal_replication>true</internal_replication>
                <replica>    <!--该分片的第一个副本-->
                    <host>hadoop102</host>
                    <port>9000</port>
                </replica>
                <replica>    <!--该分片的第二个副本-->
                    <host>hadoop103</host>
                    <port>9000</port>
                </replica>
            </shard>

            <shard>  <!--集群的第二个分片-->
                <internal_replication>true</internal_replication>
                <replica>    <!--该分片的第一个副本-->
                    <host>hadoop104</host>
                    <port>9000</port>
                </replica>
            </shard>
        </gmall_cluster>
    </clickhouse_remote_servers>

    <zookeeper-servers>
        <node index="1">
            <host>hadoop102</host>
             <port>2181</port>
        </node>
        <node index="2">
             <host>hadoop103</host>
              <port>2181</port>
        </node>
        <node index="3">
              <host>hadoop104</host>
              <port>2181</port>
        </node>
    </zookeeper-servers>

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

(2) 将hadoop102的metrika-shard.xml同步到103和104

• [root@hadoop102 config.d]# scp /etc/clickhouse-server/config.d/metrika-shard.xml root@hadoop103:/etc/clickhouse-server/config.d/metrika-shard.xml 
• [root@hadoop102 config.d]# scp /etc/clickhouse-server/config.d/metrika-shard.xml root@hadoop104:/etc/clickhouse-server/config.d/metrika-shard.xml

(3) 修改103和104中metrika-shard.xml宏的配置

##103
[root@hadoop103 ~]$ vim /etc/clickhouse-server/config.d/metrika-shard.xml
        <macros>
                <shard>01</shard>   <!--不同机器放的分片数不一样-->
                <replica>rep_1_2</replica>  <!--不同机器放的副本数不一样-->
        </macros>

##104
[root@hadoop104 ~]$ vim /etc/clickhouse-server/config.d/metrika-shard.xml
        <macros>
                <shard>02</shard>   <!--不同机器放的分片数不一样-->
                <replica>rep_2_1</replica>  <!--不同机器放的副本数不一样-->
        </macros>

(4) 在hadoop102上修改/etc/clickhouse-server/config.xml

 

(5) 同步/etc/clickhouse-server/config.xml到103和104

• [root@hadoop102 config.d]# scp /etc/clickhouse-server/config.xml root@hadoop103:/etc/clickhouse-server/config.xml
• [root@hadoop102 config.d]# scp /etc/clickhouse-server/config.xml root@hadoop104:/etc/clickhouse-server/config.xml

(6) 重启三台服务器上的ClickHouse服务

•  sudo systemctl stop clickhouse-server
•  sudo systemctl start clickhouse-server
•  ps -ef |grep click

(7) 在hadoop102上执行建表语句

•  会自动同步到hadoop103和hadoop104上
•  集群名字要和配置文件中的一致
•  分片和副本名称从配置文件的宏定义中获取

#启动客户端，并创建表
clickhouse-client -m

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_0105','{replica}')
   partition by toYYYYMMDD(create_time)
   primary key (id)
   order by (id,sku_id);

      

 可以到hadoop103和hadoop104上查看表是否创建成功

(8) 在hadoop102上创建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()

   

(9) 在hadoop102上插入测试数据

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

(10)通过查询分布式表和本地表观察输出结果

分布式表：SELECT * FROM st_order_mt_all;

  

本地表： select * from st_order_mt;

观察数据的分布