大数据-142 - ClickHouse 集群 副本和分片 Distributed 附带案例演示
点一下关注吧！！！非常感谢！！持续更新！！！
目前已经更新到了：
- Hadoop（已更完）
- HDFS（已更完）
- MapReduce（已更完）
- Hive（已更完）
- Flume（已更完）
- Sqoop（已更完）
- Zookeeper（已更完）
- HBase（已更完）
- Redis （已更完）
- Kafka（已更完）
- Spark（已更完）
- Flink（已更完）
- ClickHouse（正在更新···）
章节内容
上节我们完成了如下的内容：
- 副本和分片，上节主要是副本的部分
- Zk 的配置
- Replicated MergeTree原理详解
分片部分
副本（Replica）
- 概念：副本是指在多个节点之间复制相同的数据，确保即使某个节点发生故障，数据仍然可以从其他节点访问。副本提升了数据的高可用性和容错能力。
- 目的：如果某个节点宕机，查询可以从其他副本节点读取数据，保证系统的高可用性。同时，副本还可以均衡负载，在高并发查询时，副本节点可以分担查询压力。
- 实现：每个分片的副本节点存储相同的数据，ClickHouse 提供自动的副本同步机制，在节点恢复或新增副本时，数据会自动同步。
副本节点使用 ZooKeeper 进行协调，确保数据的写入在副本之间一致，并自动管理副本间的同步操作。 - 配置：副本通常通过 ReplicatedMergeTree 表引擎来实现，表的路径和副本 ID 由 ZooKeeper 进行管理。
Distributed 表
- 概念：Distributed 表是一种特殊的表类型，它不直接存储数据，而是将查询转发到多个分片或副本表中。这使得用户可以对多个节点执行统一的查询。
- 目的：通过 Distributed 表，可以将查询透明地分发到各个分片和副本上，最大化利用集群的并行处理能力。它简化了跨节点、跨分片查询的复杂性。
- 实现：在定义 Distributed 表时，需要指定目标集群、数据库和底层存储表的名字。
查询 Distributed 表时，ClickHouse 会根据分片键（如果存在）将查询转发到各个分片执行，并将各分片的结果汇总返回。
Distributed 表可以自动处理分片和副本的负载均衡。
分片、副本与 Distributed 表的组合
- 分片与副本的组合：通过分片，集群可以水平扩展，而通过副本，集群能够实现高可用性。当一个集群有多个分片和副本时，ClickHouse 会首先将数据分片，确保每个分片在不同的服务器上；每个分片的数据会有多个副本，副本分布在不同的节点上。
- 查询策略：查询通常会通过 Distributed 表执行。ClickHouse 会自动选择一个副本来读取数据，如果某个副本不可用，它会自动切换到其他可用副本上。查询时，可以利用并行处理，在多个分片上同时进行查询计算，提升整体查询性能。
- 副本一致性：当数据写入到副本时，ClickHouse 使用强一致性协议，确保每个副本在写入时数据是相同的。通过 ZooKeeper 管理副本的同步和协调，副本在恢复后可以从其他节点拉取丢失的数据。
优点与挑战
优点
- 高可用性：通过副本机制，即使某个节点宕机，查询和数据仍然可用。
- 可扩展性：分片机制允许系统在大规模数据场景下水平扩展。
- 高性能：Distributed 表的并行查询处理机制大大提升了查询速度，尤其在多分片、多节点的环境下。
挑战
- 管理复杂性：集群、分片、副本、ZooKeeper 之间的协调关系比较复杂，配置和维护需要较高的技术能力。
- 数据延迟：虽然副本同步机制较为强大，但在某些极端情况下，副本之间可能存在数据延迟。
配置文件
我们配置集群的时候，已经配置过了。
这里我把配置文件在粘贴到这里一次：（记得端口的事情）
<yandex>
<remote_servers>
<perftest_3shards_1replicas>
<shard>
<internal_replication>true</internal_replication>
<replica>
<host>h121.wzk.icu</host>
<port>9000</port>
<user>default</user>
<password>clickhouse@wzk.icu</password>
</replica>
</shard>
<shard>
<internal_replication>true</internal_replication>
<replica>
<host>h122.wzk.icu</host>
<port>9000</port>
<user>default</user>
<password>clickhouse@wzk.icu</password>
</replica>
</shard>
<shard>
<internal_replication>true</internal_replication>
<replica>
<host>h123.wzk.icu</host>
<port>9000</port>
<user>default</user>
<password>clickhouse@wzk.icu</password>
</replica>
</shard>
</perftest_3shards_1replicas>
</remote_servers>
<zookeeper-servers>
<node index="1">
<host>h121.wzk.icu</host>
<port>2181</port>
</node>
<node index="2">
<host>h122.wzk.icu</host>
<port>2181</port>
</node>
<node index="3">
<host>h123.wzk.icu</host>
<port>2181</port>
</node>
</zookeeper-servers>
<macros>
<shard>01</shard>
<replica>h121.wzk.icu</replica>
</macros>
<networks>
<ip>::/0</ip>
</networks>
<clickhouse_compression>
<case>
<min_part_size>10000000000</min_part_size>
<min_part_size_ratio>0.01</min_part_size_ratio>
<method>lz4</method>
</case>
</clickhouse_compression>
</yandex>
Distributed用法
Distributed表引擎
- all 全局查询
- local 真正的保存数据的表
Distributed
分布式引擎，本身不存储数据，但可以在多个服务器上进行分布式查询。
读是自动并行的，读取时，远程服务器表的索引（如果有的话）会被使用。
指令是：
Distributed(cluster_name, database, table [, sharding_key])
参数解析：
- cluster_name 服务器配置文件中的集群名，在我们配置的metrika.xml中
- database 数据库名
- table 表名
- sharding_key 数据分片键
案例演示
创建新表
在3台节点上的t表插入新建表
注意：是三台节点都要！！！
CREATE TABLE test_tiny_log(
id UInt16,
name String
) ENGINE = TinyLog;
执行结果如下图所示：
插入数据
在3台节点上的t表插入一些数据
注意：是三台节点都要！！！
INSERT INTO test_tiny_log VALUES (1, 'wzk');
INSERT INTO test_tiny_log VALUES (2, 'icu');
SELECT
*
FROM
test_tiny_log
执行结果如下图所示：
分布式表
CREATE TABLE dis_table(
id UInt16,
name String
) ENGINE = Distributed(perftest_3shards_1replicas, default, test_tiny_log, id);
执行代码如下：
插入数据
INSERT INTO dis_table SELECT * FROM test_tiny_log;
插入我们刚才准备的数据：
查询数据
select count() from dis_table;
运行结束后，对应的截图如下所示：
查看每台节点的数据：
SELECT COUNT() FROM test_tiny_log;
执行结果如下图：
h121节点的返回：
h122节点的返回：
h123节点的返回：
可以看到三台的总数量（2 + 3 + 3）等于我们的分布式表dis_table（8）的数量，每个节点大约有 1/3 的数据。