ClickHouse 云原生使用手册

k8s技术圈

ClickHouse 云原生使用手册

背景

为适应越来越大的实时 OLAP 需求，我们引入了 ClickHouse。在探索 ClickHouse 和云原生结合的路上，我们逐渐总结出一套 ClickHouse 的使用方案，该方案有以下特点：

高性能

ClickHouse 使用 C++ 语言编写，引入向量化执行引擎，在存储和计算方面进行了大量的优化，是目前业内性能最高的 OLAP 引擎。查询分析类计算比 MySQL 快 100 倍以上，更详细的性能对比可查看 ClickHouse 官方的 benchmark

低成本

ClickHouse 支持冷热分离的存储策略，我们将热数据存储在 EBS 上，将大部分冷数据存储在 S3 上，获得了不限量的低成本存储

可扩展

ClickHouse 支持计算资源的水平扩展，我们将 ClickHouse 部署在可以伸缩的 Nomad Clients 上，并通过使用 S3 获得了不限量的存储空间

稳定可靠

ClickHouse 的 ReplicatedMergeTree 引擎支持通过副本表的形式将数据冗余地存储在多个节点上，来提高可用性，同时我们引入了 Coodinator Node 来保证各个节点上计算任务均衡，保证我们的集群面对并发的场景仍然有比较可靠的可用性

方案

本方案由两组 ClickHouse 进程组成，分别为负责协调和调度读写操作的 Coodinator Node，与负责数据读写的 Data Node 组成，命名的方式参考了 ElasticSearch

（ClickHouse 云原生部署架构图）

此架构有如下几个特点：

1. 存在一组可水平扩展的 Data Node，用来执行数据的读、写操作

2. 存在一组可水平扩展的 Coodinator Node，扮演着代理的角色，所有通过 Data Node 进行的数据操作，包括建表（DDL）、对数据的读写都需要通过 Coodinator Node 进行

3. Coodinator Node 和 Data Node 都依赖同一组 ZooKeeper 服务进行协调

4. 因为 ClickHouse 支持通过 TCP 进行连接的 Client，所以我们通过 NLB 对 Coodinator Node 的连接进行负载均衡

5. Data Node 既包含 EBS 的存储也包含 S3 的存储，通过冷热分离的功能，把比较久远的数据放在 S3 上以缩减成本

接下来我们会详细介绍我们使用的集群模式，各个节点的作用以及使用的配置项

Data Node

Data Node 是集群中负责数据存储的节点，会存储数据表等元数据，也负责执行数据的读、写以及 JOIN 等操作。集群中的所有 Data Node 可以按照 cluster 和 shard 分成若干组。

（cluster、shard 和 data node 的关系）

我们通过 ReplicatedMergeTree 这一 Engine 在同一个 shard 的不同 Data Node 之间同步数据（具体原理下文会详细描述）以保证数据和元数据是冗余存储的，不会因为单点问题而丢失。

Data Node 中包含集群的配置，用于执行分布式 SQL 语句，Data Node 间的数据复制不依赖此配置

YAML
remote_servers:
clickhouse_dev_data:
    shard:
      - replica:
          host: 127.0.0.1
          port: 9000
      - replica:
          host: 127.0.0.1
          port: 9000

Data Node 间的数据复制主要依赖各个 Data Node 在 Zookeeper 中注册的 hostname 进行通信。hostname 可通过 interserver_http_host 配置。如果不配置使用和 hostname -f 一致的名称。

Coordinator Node

为了保证同一个 shard 中的多个 Data Node 上的计算任务是均衡的，我们引入了 Coordinator Node 作为一层代理。在 Coodinator Node 上建立分布式表，即可通过 Coodinator Node 去查询 Data Node 所管理的数据，Coodinator Node 对于查询或者写入的 Client 来说是 Server，对 Data Node 来说是 Client。

长期来看，分布式表的元数据会保存在 Coodinator Node 的磁盘上，因此这一层代理是有状态的，只有正确的建立分布式表才能满足查询的要求。同时，元数据所占用的磁盘空间也比较小，所以 Coodinator Node 需要的 EBS Volumn 的 size 也会比较小。

Coodinator Node 和 Data Node 之间也需要鉴权，因此 Coodinator Node 的配置文件中需要填写 Data Node 的鉴权信息。当发生读/写操作时，Coodinator Node 会直连 Data Node 进行操作，具体流程下文会详细描述。

YAML
remote_servers:
clickhouse_data:
    shard:
      # 需要配置用户名，密码来连接 Data Node
      - replica:
          host: 127.0.0.1
          port: 9000
          user: default
          password: xxx
      - replica:
          host: 127.0.0.2
          port: 9000
          user: default
          password: xxx
clickhouse_coordinator:
    shard:
      - replica:
          host: 127.0.0.3
          port: 9000
       - replica:
          host: 127.0.0.4
          port: 9000

通用配置

集群内的节点间同步 DDL 语句，依赖 ZooKeeper，相关配置如下：

YAML
# ZooKeeper 连接信息
zookeeper:
- node:
host: zookeeper.service.consul
port: 2181

# 监听 ZooKeeper 的此目录获取分布式 DDL 的信息
distributed_ddl:
path: /clickhouse/task_queue/ddl

存储

在 ClickHouse 中，我们根据不同类型表引擎来选择不同的存储介质。日常中我们主要使用 MergeTree 系列引擎以及 S3 引擎，下面我们主要介绍这 2 种引擎。

MergeTree 引擎

MergeTree 系列引擎主要负责 ClickHouse 的数据存储和查询，是 ClickHouse 最核心的表引擎，该引擎支持我们使用 EBS 和 S3 共同作为 ClickHouse 的存储介质。在实际使用场景中我们通常按数据的存储周期采用冷热分离的策略，将热数据存储在 EBS 上，将冷数据存储在 S3 上。这样既能够保证热数据的查询和写入性能，又能降低冷数据的存储成本。

EBS 存储配置可参考：Path
S3 存储的配置可参考：Using S3 for Data Storage
冷热分离数据移动策略可参考：Using Multiple Block Devices for Data Storage

当我们同时使用 EBS + S3 作为存储介质时的配置如下：

YAML
# EBS 存储路径
path: /data/clickhouse

storage_configuration:
disks:
    # S3 访问相关信息
    s3:
      type: s3
      endpoint: https://test-dev-clickhouse.s3.cn-northwest-1.amazonaws.com.cn/clickhouse/
      # 使用机器级别的 IAM Role
      use_environment_credentials: true
      region: cn-northwest-1
policies:
    # 数据从 EBS 移动到 S3 的策略
    move_from_ebs_to_s3:
      volumes:
        main:
          disk: default
        external:
          disk: s3
      move_factor: 0.2

在上述配置文件中，我们定义了一个名为 move_from_ebs_to_s3 的存储策略，并在存储策略中添加 2 种不同类型的 volume。其中名为 main 的 volume 将数据存储在 disk 为 default 块存储上（即本地 EBS 中），名为 external 的 volume 则将数据存储在 disk 为 s3 的对象存储上。

当往 ClickHouse 写入数据时，ClickHouse 会优先写入 volumes 列表中的第 1 个 volume（本例子中是 main）当本 volume 的可用存储空间低于阈值时，则本 volume 中的数据以 part 为单位移动到下一个 volume 中。

volume 的详细配置如下：

max_data_part_size_bytes : 单个 part 的最大大小。如果超过，则直接将该 part 转移至 S3。
move_factor : 当磁盘的可用空间量低于此因子时，数据会自动迁移至下一个 volume。默认为 0.1，即 90%。
prefer_not_to_merge : 禁止某个 volume 上的 part 合并
perform_ttl_move_on_insert : 默认情况下，插入已经过期的数据时，会自动直接移动下一个存储中。这可能会降低写入的速度。建议 Disabled
load_balancing : 写入时 volume 的平衡策略。我们暂时用不上。

S3 引擎

S3 引擎主要用于查询 AWS S3 上已存在的数据，属于 ClickHouse 扩展功能。当我们使用 S3 引擎时，我们需要在配置文件中做如下声明：

YAML
s3:
test-dev-analytics:
endpoint: https://test-dev-analytics.s3.ap-east-1.amazonaws.com/
use_environment_credentials: true

在上面配置中，我们声明了一个名为shiyou-labs-aws-cur-test的 S3 endpoint 点并配置 endpoint 访问信息。我们声明 use_environment_credentials 为 true 来让 ClickHouse 通过机器级别的 IAM Role 的方式来访问 S3。该方式需要 S3 服务为提供 ClickHouse 服务的 EC2 所关联的 IAM Role（后面将该 IAM Role 称为 ClickHouse IAM Role）授予访问权限。

建表

分布式 DDL

默认情况下，我们在节点上执行 CREATE、ALTER、DROP、RENAME 及 TRUNCATE 这些 DDL 语句时，该语句仅仅只能在本地节点生效。并不能在整个集群所有节点上生效，如果要想在整个集群的所有节点上生效，则默认需要分别登录每个节点上面手动执行 CREATE 等这些 DDL 语句。这显然是繁琐也容易出错的。

ClickHouse 为我们提供了分布式 DDL 功能。该功能可以允许我们在一台节点上使用 on cluster {cluster} 语法执行 DDL 语句，从而 ClickHouse 帮我们自动分发到集群中的所有节点上执行，省去了需要依次去单个节点执行 DDL 的烦恼。

分布式 DDL 语句的执行依赖 Zookeeper 做协同，相关配置如下：

YAML
zookeeper:
- node:
      host: zookeeper.service.consul
      port: 2181

distributed_ddl:
path: /clickhouse/task_queue/ddl

remote_servers:
clickhouse_dev:
    shard:
      - internal_replication: true
        replica:
          host: 10.0.0.0
          port: 9000
      - internal_replication: true
        replica:
          host: 10.0.0.1
          port: 9000

分布式 DDL 语句如下：

SQL
CREATE TABLE default.test on cluster clickhouse_dev
(
    `id` Int64 COMMENT '主键ID',
    `name` String COMMENT '名称',
    `create_time` DateTime COMMENT '创建时间',
    `dt` String DEFAULT toYYYYMMDD(create_time) COMMENT '分区字段',
)
ENGINE = MergeTree()
PARTITION BY dt
ORDER BY (id, create_time)

ClickHouse 分布式 DDL 用于在集群内的多个 Data Node 上批量执行 DDL 语句，以简化我们在多 Data Node 上逐个执行 DDL 的操作。整个分布式 DDL 的执行流程如下：

客户端通过 on cluster 关键字向 Coordinator Node 发起分布式 DDL 请求，用于在 Coordinator Node 上向集群中所有 Data Node 分发执行 DDL 语句
由于 Coordinator Node 并没有使用 Zookeeper 对集群中的 Data Node 做服务发现，故 Coordinator Node 需要在 system.clusters 表中查询集群中所有的 Data Node 节点
Coordinator Node 将 DDL 操作日志推送到 Zookeeper 上的 ddl 节点/clickhouse/task_queue/ddl/query-000000207 并监控执行进度。推送到 Zookeeper 上操作日志样例数据如下：

YAML

[zk: localhost:2181(CONNECTED) 4] get /clickhouse/task_queue/ddl/query-0000000207
version: 2
query: CREATE TABLE default.test_distributed_ddl UUID \'4355ae32-b5c9-4a78-a4c7-7e3b7b333beb\' ON CLUSTER clickhouse_dev (`id` Int64 COMMENT \'主键ID\', `text` String COMMENT \'文本字段\', `create_time` DateTime COMMENT \'创建时间\', `dt` String DEFAULT toYYYYMMDD(create_time) COMMENT \'分区字段\') ENGINE = ReplicatedMergeTree(\'/clickhouse/tables/{shard}/{database}/{table}\', \'{replica}\') PARTITION BY dt ORDER BY (id, create_time) SETTINGS storage_policy = \'move_from_ebs_to_s3\'
hosts: ['10%2E0%2E0%2E22:9000','10%2E0%2E0%2E23:9000']
initiator: ip%2D10%2D0%2D0%2D26%2Eap%2Deast%2D1%2Ecompute%2Einternal:9000
settings: load_balancing = 'random', max_memory_usage = 10000000000

可以看到操作日志中主要包含：

version：版本号
query：需要分发执行的 DDL 语句
hosts：需要执行 DDL 语句的 Data Node 节点列表
initiator：操作日志推送者
setting：

load_balancing：为多副本查询时的负载均衡算法

max_memory_usage：单节点的内存使用限制

4. Data Node 监听 Zookeeper 上 ddl 节点下的变化并获取操作日志。通过 C++ 的 ifaddrs 包获取本机 IP 地址信息并判断本机 IP 是否在 hosts 列表中。如果在，则执行 query 中的 DDL 语句。如果不在，则直接跳过即可。

5. Data Node 执行完 query 中的 DDL 语句后，会将本节点的状态写入到 finished 节点中，finished 节点中的样例数据如下：

YAML

[zk: localhost:2181(CONNECTED) 4] ls /clickhouse/task_queue/ddl/query-0000000207/finished
[10%2E0%2E0%2E22:9000, 10%2E0%2E0%2E23:9000]

如上可以看到，执行完成的节点都会在 finished 节点下创建对应的子节点。

6. Coordinator Node 监听 finished 节点下子节点的变化来判断集群中的所有节点是否全部执行完成。整个监听过程是同步阻塞的。如果全部执行完成则返回给客户端，如果没有全部执行完成，则默认阻塞 180s 后超时失败。阻塞超时时间可以通过distributed_ddl_task_timeout 来调整。

7. 整个流程完毕后，Coordinator Node 将结果返回给客户端

在执行分布式 DDL 时，我们需要注意以下几点：

整个分布式 DDL 流程中依赖 Zookeeper 为协调，故需要保证 Zookeeper 的可用性。否则 Coordinator 无法发起分布式 DDL 执行流程
整个分布式 DDL 依赖 Zookeeper 上指定的节点（distributed_ddl.path）进行任务的扭转和监听，则 Coordinator Node 和 Data Node 节点上的distributed_ddl.path 配置必须一致。否则 Coordinator Nod
往 Zookeeper 上写入 task，Data Node 无法监听并执行，则整个分布式 DDL 堵塞到 180s 后会自动超时。

在集群模式下，ClickHouse 的建表方式和其他数据库存在一些差异。我们需要在 Coordinator Node 和 Data Node 上面创建不同的表引擎，才能完成集群模式下的数据写入和查询。整个建表流程如下：

在 Data Node 上创建数据表

ClickHouse MergeTree 系列的表引擎是 ClickHouse 数据存储功能的核心，它们为弹性和高性能数据检索提供了大多数功能：列存储、自定义分区、一级索引、二级索引等。为了保证数据的高可用性，我们选择在 Data Node 上使用 ReplicatedMergeTree 引擎创建表。整个建表流程如下：

1. 客户端使用 on cluster 关键字向 Coordinator Node 发送分布式 DDL 建表请求，建表语句如下：

SQL
CREATE TABLE default.test on cluster {cluster}
(
    `id` Int64 COMMENT '主键ID',
    `name` String COMMENT '名称',
    `create_time` DateTime COMMENT '创建时间',
    `dt` String DEFAULT toYYYYMMDD(create_time) COMMENT '分区字段',
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}','{replica}')
PARTITION BY dt
ORDER BY (id, create_time)
TTL create_time + INTERVAL 1 HOUR TO DISK 's3'
SETTINGS storage_policy = 'move_from_ebs_to_s3';

如上所示，MergeTree 中建表语句的参数如下：

1. cluster：ClickHouse 的集群名称。

2. shard: Data Node 所属的 shard 编号。该参数自动从系统表中获取

3. database：库名。该参数自动从系统获取

4. table：表名。该参数自动从系统获取

2. Coordinator Node 节点通过集群名从 system.clusters 表中获取该集群对应的所有 Data Node 节点信息

3. Coordinator Node 将 DDL 语句发送到 Zookeeper 节点中

4. 集群中的各个 Data Node 从 Zookeeper 中获取 DDL 语句执行

至此，所有 Data Node 上的本地表创建结束。由于集群模式下，我们只将 Coordinator Node 暴露给客户端访问，此时我们在 Data Node 上创建的表还不能被客户端访问到。故我们接下来需要在 Coordinator Node 上建表。

在 Coordinator Node 上建表

ClickHouse 的 Distributed 表引擎支持数据写入和查询的转发，本身不存储任何数据。故我们在 Coordinator Node 上使用 Distributed 引擎创建表，用于对 Data Node 上表的访问。

具体建表流程如下：

1. 客户端向 Coordinator Node 发送建表请求

具体的建表语句如下：

SQL
CREATE TABLE default.test_distributed
(
    `id` Int64 COMMENT '主键ID',
    `name` String COMMENT '名称',
    `create_time` DateTime COMMENT '创建时间',
    `dt` String DEFAULT toYYYYMMDD(create_time) COMMENT '分区字段',
)
ENGINE = Distributed('{cluster}','{database}','{table}');

如上建表语句，Distributed 引擎中参数含义如下：

cluster：对应 Coordinator Data 配置中 Data Node 集群的名称
database：需要查询的 Data Node 节点上的库名
table：需要查询的 Data Node 节点上的表名
Coordinator Node 在本地执行建表语句

至此，我们整个建表流程介绍。

注意：

1. 当我们执行 drop table 删除 ReplicatedMergeTree 表引擎的时候，会存在无法立即删除ReplicatedMergeTree 存在于 zookeeper 上的元数据信息。此时我们需要在 drop table 后添加 no delay 参数去解决。官方 issue ReplicatedMergeTree in Atomic database。删除语句如下：

SQL

DROP TABLE default.test_ttl ON cluster clickhouse_dev no delay;

分布式写入

接下来我们将描述集群模式下，如何写入数据，大致分为以下几步：

1. Client 向 Coodinator Node 发送要写入的数据

2. Coodinator Node 本地缓存该数据并开始向 Data Node 推送

3. 完成写入的 Data Node 在同一个 shard 内进行数据同步

（写入时序图）

如上所示通过 Distributed 表引擎进行分布式写入的步骤如下：

1. Coordinator Node 上的 Distributed 引擎接收到数据后会先写入本地临时目录，例如：/{data}/{database}/{table}/{shard_number}/tmp

2. Coordinator Node 定时监听临时目录中的数据变化，有变化时触发向 Data Node 推送数据的功能，定时监听的频率涉及如下 2 个参数配置：

1distributed_directory_monitor_sleep_time_ms ：基本间隔频率，默认为 100毫秒

1distributed_directory_monitor_max_sleep_time_ms ：最大间隔频率，默认为 30秒

3. 默认情况下 Coordinator Node 会在每个 shard 中挑选一个 Data Node 逐个推送已写入的数据文件。文件推送涉及如下 2 个参数配置：

3.1. distributed_directory_monitor_batch_inserts：开启文件的批量发送，默认为禁止

3.2. background_distributed_schedule_pool_size：发送文件的线程池大小，默认为 16。只能在服务启动时设置

4. Data Node 1 接收 Coordinator 发送的数据文件并写入本地

5. 同一个 shard 的多个 Data Node 1 之间进行数据复制

整个数据写入流程有同步和异步 2 种模式，可通过 insert_distributed_sync 进行配置：

5.1 异步，默认方式，Coordinator Node 将数据写入到本地临时目录后，就会向客户端返回写入成功信息。

5.2 同步，Coordinator Node 必须保证数据写入至少 1 个 Data Node 节点上时，整个 Insert 操作才会成功。

由于异步模式无法在 Coodinator Node 宕机时保证数据一定能写入 Data Node，所以我们此处会使用同步模式。

Data Node 之间的数据复制

当 shard 中的某一个 Data Node 成功完成了本地的写入操作后，ClickHouse 会通过 ReplicatedMergeTree 来 shard 内部的数据同步。数据复制过程依赖 Zookeeper 做协同，官方建议使用的 Zookeeper 版本最好是 3.4.5 及以上。

包含副本状态感知、操作日志分发、任务队列等。同一个 shard 的不同副本只能通过配置文件的方式指定 IP 和端口。

下面我们以 2 个 Data Node 为例，来介绍 Data Node 间的数据复制流程

（数据复制时序图）

1. Data Node 1 从 Coodinator Node 接收数据，并将数据按条数（max_insert_block_size = 1048576）切分成若干个 Block 块并写入本地分区。

2. Data Node 1 将操作日志写入到 Zookeeper 的 log 路径下

（/clickhouse/tables/{shard}/{database}/{table}/log/log-00000197）具体样例数据如下：

YAML

[zk: localhost:2181(CONNECTED) 40] get /clickhouse/tables/1/default/test_ttl/log/log-0000000197
format version: 4
create_time: 2022-09-27 08:00:29
source replica: replica_2
block_id: f880fa005b8940dcda64418c7eb81b52_1332684842122574982_11562910527007783636
get
f880fa005b8940dcda64418c7eb81b52_61_61_0

可以看到操作日志包含如下内容：

2.1 create_time：Log 数据写入 Zookeeper 的时间

2.2 source replica：操作日志的推送副本

2.3 block_id：当前 part 的 Block ID

2.4 type：操作类型。

2.5 part id：当前分区的目录 ID

2.6 Data Node 1 更新本节点在 Zookeeper 下的 log_pointer 下标。具体路径为：/clickhouse/tables/{shard}/{database}/{table}/replicas/{replica_name}/log_pointer。

3. Data Node 2 监听 Zookeeper log 路径下的节点变化，并从 log pointer 中获取最新的 log 点。然后将 log pointer 点往后的操作日志转化为 task 任务。并放到 Zookeeper 上的 queue 队列中（/clickhouse/tables/{shard}/{database}/{table}/replicas/{replica_name}/queue）

4. Data Node 2 从 queue 队列中获取任务并选择一个最佳的副本节点去拉取数据。副本选择算法如下：

4.1 获取 replicas/下的所有副本并通过 is_active仅保留活跃的副本

4.2 获取每个副本的 log_pointer，该值代表已同步的 part 下标。值越大则该节点的数据越齐全

4.3 获取每个副本的 queue，该值代表为同步的 part。值越大则该节点的数据越少

4.4 选择log_pointer最大的副本，如果 log_pointer一致，则选取 queue 长度越小的副本

5. Data Node 2 通过 Zookeeper 上的 replicas/{replica_name}/host获取副本的 hostname 信息。该 hostname 可通过 interserver_http_host 配置。如果不配置则和 hostname -f 一致。样例数据如下：

YAML

[zk: localhost:2181(CONNECTED) 0] get /clickhouse/tables/1/default/test_ttl_2/replicas/replica_1/host
host: ip-10-0-0-12.ap-east-1.compute.internal
port: 9009
tcp_port: 9000
database: default
table: test_ttl_2
scheme: http

6. 向远端副本发送下载请求，如果下载失败，则尝试 5 次。具体可通过 max_fetch_partition_retries_count 进行配置。

默认情况下，数据成功写入 1 个副本后就会返回给客户端，其他副本异步同步。当该副本所在的节点宕机时，则可能会导致数据丢失。我们可以通过 insert_quorum 参数来保证多副本同时写入成功。

这样会损失一些写入性能，与非复制表相比，此操作会增加写入的延迟。官方建议我们以低于 1 秒的频率批量写入数据，这部分损耗是可以忽略不计的。

分布式查询

单表查询

1. 客户端向 Coordinator Node 上的 Distributed 表发送查询请求

2. Coordinator Node 通过负载均衡算法为每个 Shard 选择一个 Data Node 副本节点。负载均衡算法通过 load_balancing 配置指定：

3. Coordinator Node 将请求分发给被选中的 Data Node

4. Data Node 将查询结果返回给 Coordinator Node

5. Coordinator Node 将结果返回客户端

普通 JOIN

从上面的流程图我们可以看出 ClickHouse 中的 join 流程如下：

1. 客户端向 Coordinator Node 上的分布式表发送查询请求（select name from a join b where a.id = b.id）

2. Coordinator Node 上的分布式表通过负载均衡算法，为每个 shard 选择 1 个副本节点。目前我们的集群属于 1 个 shard 2 个副本的架构

3. Coordinator Node 将查询发送到 a 表被选中的 Data Node

4. 由于 a 表需要和 b 表做 join，则 a 表所在的 Data Node 需要向 b 表所在的所有 Data Node 发送查询请求并将 b 表的数据拷贝到本地做 join。如果 a 表的数据分布在 M 个 Data Node 上，b 表的数据分布到 N 个 Data Node 上。则在整个 join 过程中，a 表需要发送 M * N 个数据查询请求。

5. 每个 a 表所在的 Data Node 将 join 后的数据返回给 Coordinator Node 上的分布式表

6. Coordinator 上的分布式表将各个 Data Node 返回的数据合并

7. Coordinator 上的分布式表将数据返回给客户端

从第 4 步可以看出当 2 个分布式表进行 join 时，会产生大量的数据查询请求（shard 数的平方），由于我们的集群目前是 1 个shard 2 个副本。所以暂时不会该问题。但我们下面介绍一下解决这个问题的 Global join 的方案。

Global JOIN

从上面的流程图可以看出 ClickHouse 的 global join 如下：

1. 客户端向 Coordinator Node 上的分布式表发送查询请求（select name from a global join b where a.id = b.id）

2. Coordinator Node 上的分布式表通过负载均衡算法，为每个 shard 选择 1 个副本节点。目前我们的集群属于 1 个 shard 2 个副本的架构

3. Coordinator Node 上的分布式表先查询 b 表的数据并存储在本地临时内存表中

4. Coordinator Node 将 b 表的数据发送到 a 表所在的 Data Node（如果 a 表是副本，则会在多个副本中选择一个 Data Node）

5. Coordinator Node 将查询语句发送至 a 表所在的 Data Node

6. Data Node 将本地的 a 表及内存中的 b 进行 join

7. Data Node 将本地计算完的数据返回给 Coordinator Node

8. Coordinator Node 合并所有 Data Node 返回的数据并最终返回给客户端

在 global join 中，可以看到对 b 表的查询请求只在 Coordinator Node 上发生了一次，即请求数为 shard 数。但使用 global join 时也需要注意以下几点：

1. 一般建议对维表进行 join 的时候使用，数据量小。

2. Global join 的 Coordinator Node 到 Data Node 的网络带宽无法限制。

认证

如上所示，我们只将 Coordinator Node 暴露给客户端，由于客户端到 Coordinator Node 认证信息无法直接穿透到 Coordinator Node 到 Data Node 中去，故我们整个 ClickHouse 的认证过程分为如下 2 个阶段：

1. 客户端连接 Coordinator Node 时，通过指定不同的用户名和密码来访问不同的表。创建用户的样例语句如下：

SQL
CREATE USER test_user IDENTIFIED WITH sha256_password BY 'test_password' HOST LIKE '10.%.%.%';
grant SELECT, INSERT, ALTER, CREATE TABLE, CREATE VIEW, CREATE DICTIONARY, OPTIMIZE ON default.* to test_user;

2. Coordinator Node 连接 Data Node 时，通过预先在 Coordinator 节点中配置的用户名及密码来访问 Data Node 上面的库和表。配置方式如下：

YAML
remote_servers:
clickhouse_dev:
    shard:
      - internal_replication: true
        replica:
          host: 127.0.0.1
          port: 9000
          user: test_user
          password: test_password
      - internal_replication: true
        replica:
          host: 127.0.0.1
          port: 9000
          user: test_user
          password: test_password

最佳实践

创建用户并授权

SQL
-- 创建用户的例子
CREATE USER test_user IDENTIFIED WITH sha256_password BY 'test_password' HOST LIKE '10.%.%.%';

-- 向用户授权的例子
GRANT SELECT, INSERT, ALTER, CREATE TABLE, CREATE VIEW, CREATE DICTIONARY, OPTIMIZE ON test_database.* to test_user;

建表

MergeTree

创建 MergeTree 引擎表需要分别在 Coordinator Node 上创建 Distributed 表以及在 Data Node 上创建 MergeTree 引擎表：

1在 Coordinator Node 上执行分布式 DDL 来在 Data Node 节点上创建 MergeTree 引擎表

SQL
CREATE TABLE IF NOT EXISTS omni_sdk.app_store_sale ON cluster clickhouse_dev(
    account String COMMENT '',
    vendor_number String COMMENT '',
    dt Date COMMENT '时间分区',
    provider String COMMENT '提供商',
    provider_country String COMMENT '提供商所在国家或地区',
    sku String COMMENT 'sku',
    create_time DateTime64(9, 'UTC') COMMENT ''
)
ENGINE = ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/{database}/{table}','{replica}',create_time)
ORDER BY(account,
    vendor_number,
    dt,
    provider,
    provider_country,
    sku
)
TTL toDateTime(create_time) + INTERVAL 30 DAY TO DISK 's3'
SETTINGS storage_policy = 'move_from_ebs_to_s3'

关于建表语句的说明如下：

1. ON cluster clickhouse_dev 将建表语句分发到集群名为 clickhouse_dev 的节点上去。

2. ReplicatedReplacingMergeTree 引擎主要用于表的副本及去重功能

3. storage_policy = 'move_from_ebs_to_s3' 用于数据冷热分离

4. TTL toDateTime(create_time) + INTERVAL 30 DAY TO DISK 's3'将 30 天以前的数据转移到 S3。

1在 Coordinator Node 上创建分布式表

SQL
CREATE TABLE IF NOT EXISTS omni_sdk.app_store_sale(
    account String COMMENT '',
    vendor_number String COMMENT '',
    dt Date COMMENT '时间分区',
    provider String COMMENT '提供商',
    provider_country String COMMENT '提供商所在国家或地区',
    sku String COMMENT 'sku',
    create_time DateTime64(9, 'UTC') COMMENT ''
)
ENGINE = Distributed('clickhouse_dev','omni_sdk','app_store_sale');

S3 外表

针对 S3 外表，我们需要分别在 Data Node 和 Coordinator 节点上创建。其中 Coordinator Node 上的外表主要用于预览 S3 上的数据，Data Node 上的外表主要用于做数据的 join。

1. 我们在 Data Node 集群节点上的 S3 外表建表语句如下：

SQL
-- 创建 cur 外部表，并跳过第1列
CREATE TABLE default.test_s3_engine on cluster clickhouse_data_dev(
    identity_lineitemid String COMMENT '',
    identity_timeinterval String COMMENT '',
    bill_invoiceid String COMMENT ''
) engine = S3('https://shiyou-labs-aws-cur-test.s3.ap-southeast-1.amazonaws.com/*.csv.gz', 'CSV')
SETTINGS input_format_csv_skip_first_lines = 1;

2. 我们在 Coordinator Node 集群节点上的 S3 外表建表语句如下：

SQL
-- 创建 cur 外部表，并跳过第1列
CREATE TABLE default.test_s3_engine on cluster clickhouse_coordinator_dev(
    identity_lineitemid String COMMENT '',
    identity_timeinterval String COMMENT '',
    bill_invoiceid String COMMENT ''
) engine = S3('https://shiyou-labs-aws-cur-test.s3.ap-southeast-1.amazonaws.com/*.csv.gz', 'CSV')
SETTINGS input_format_csv_skip_first_lines = 1;

架构演进

单点架构

使用单个 EC2 部署 ClickHouse 服务，元数据存储在 EBS 上，数据存储在 S3 上。定时对 EBS 上的数据做 snapshot。特点如下：

1. 部署简单

2. 单个 EBS 具有 99.9% 的可用性，元数据的安全性可以得到较高的保证

3. 针对单个 EBS 做 snapshot，可恢复一部分人为误操作删除的数据

4. 单个 EC2 的性能无法满足业务需求时，我们可以先通过升级 EC2 配置解决

Replicated 副本表架构

Replicated 架构在单点架构的基础上通过使用 ReplicatedMergeTree 引擎来将同一份数据实时备份到多个 EC2 部署的 ClickHouse 节点上。每个节点都存储全量数据。每个服务的元数据存储在各自的 EBS 上，数据在 S3 上存储一份。相较于单节点的优势如下：

1. 多个节点 EBS 的可用性要高于单个节点

2. ClickHouse 服务可以实现不停机的滚动更新

3. 能提高查询的并行度，但无法提高单个查询效率

分布式架构

分布式架构在单点架构的基础上通过使用 Distributed 表引擎，来将写入到集群中的数据均匀的分配到多个 EC2 部署的 ClickHouse 节点上（将数据水平切分）。每个节点只存储集群中的部分数据。可以解决单节点遇到的如下问题：

1. 当单个节点的性能无法通过升级节点配置解决时，则我们可以使用该架构将数据存储在多个节点上进行并行处理。