大数据-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>[email protected]</password></replica></shard><shard><internal_replication>true</internal_replication><replica><host>h122.wzk.icu</host><port>9000</port><user>default</user><password>[email protected]</password></replica></shard><shard><internal_replication>true</internal_replication><replica><host>h123.wzk.icu</host><port>9000</port><user>default</user><password>[email protected]</password></replica></shard></perftest_3shards_1replicas></remote_servers><zookeeper-servers><nodeindex="1"><host>h121.wzk.icu</host><port>2181</port></node><nodeindex="2"><host>h122.wzk.icu</host><port>2181</port></node><nodeindex="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表插入新建表
注意：是三台节点都要！！！

CREATETABLE test_tiny_log(
  id UInt16,
  name String
)ENGINE= TinyLog;

执行结果如下图所示：

插入数据

在3台节点上的t表插入一些数据
注意：是三台节点都要！！！

INSERTINTO test_tiny_log VALUES(1,'wzk');INSERTINTO test_tiny_log VALUES(2,'icu');SELECT*FROM
  test_tiny_log

执行结果如下图所示：

分布式表

CREATETABLE dis_table(
  id UInt16,
  name String
)ENGINE=Distributed(perftest_3shards_1replicas,default, test_tiny_log, id);

执行代码如下：

插入数据

INSERTINTO dis_table SELECT*FROM test_tiny_log;

插入我们刚才准备的数据：

查询数据

selectcount()from dis_table;

运行结束后，对应的截图如下所示：

查看每台节点的数据：

SELECTCOUNT()FROM test_tiny_log;

执行结果如下图：
h121节点的返回：

h122节点的返回：

h123节点的返回：

可以看到三台的总数量（2 + 3 + 3）等于我们的分布式表dis_table（8）的数量，每个节点大约有 1/3 的数据。