本文介绍 ClickHouse 集群的基本使用操作。

1 前提条件

已创建火山引擎 E-MapReduce（EMR）包含 ClickHouse 的集群类型。详见快速开始。

2 环境介绍

1. 登录 EMR 控制台。

2. 单击集群列表 > 服务列表 > ClickHouse > 部署拓扑页签，进入 ClickHouse 组件服务的部署拓扑。

3. 单击组件名称下 (emr-core-1 主机名称)的 ECS ID，跳转进入到云服务器的实例界面，点击右上角的远程连接按钮。

   

4. 选择一种远程连接方式（推荐选择 ECS Terminal），并输入集群相关认证信息，登录到 ClickHouse 集群的 Core 实例节点命令行环境中，以执行相关命令行操作。

5. 通过客户端连接 ClickHouse 集群

   /usr/lib/emr/current/clickhouse/clickhouse-client
   

2.1 查看集群信息

select cluster,shard_num,replica_num,host_address from system.clusters;

结果如上图所示，ClickHouse 默认的 Cluster name 为 cluster_emr，该示例集群有 3个 shard，每个 shard 只有一个副本。

2.2 查看集群宏信息

select * from system.macros;

在示例集群中有三个宏: cluster，replica，shard，这三个宏是 EMR 集群创建后，自动会加上。
集群宏可以认为是节点上的变量，例如宏 shard，在这个节点上的值为 1，在另外一个节点上的值就会是2或者3。
这里我们需要记住这个 cluster_name，以及这三个宏的名字，后续过程会被使用到。

3 创建数据库

ClickHouse 有多种 Database Engine，具体的区别可以参考官方文档，在下面的示例中，会按照默认的 Atomic Engine 来举例。

3.1 创建单机数据库

执行以下 SQL 来创建一个名为 test 的数据库

CREATE DATABASE test ENGINE = Atomic;

创建完成后，执行show databases就能看到有个 test 的数据库。
但是，需要注意的是，上面的 SQL 只会在当前的节点创建数据，其他 shard 或者 replica 都不会创建 test 的数据库。
因此这条 SQL 只是一条单机的 DDL 语句。

3.2 分布式创建数据库

分布式 SQL 需要特殊的建表语句，对于创建数据库，我们需要执行如下 SQL:

CREATE DATABASE test ON CLUSTER cluster_emr  ENGINE = Atomic;

执行 SQL 后，ClickHouse 客户端会反馈每个节点的执行情况。 如上图所示，已经在示例集群的3个节点中全部执行完成。
分布式 DDL 的句式就是原本单机的 DDL 后，追加一句 "ON CLUSTER {cluster_name}" 的语法，在 EMR 的集群中， {cluster_name}为 cluster_emr
更多关于分布式 DDL 的使用方式，可以参考官方文档。

4 创建多副本表

ClickHouse 有多种表引擎，在生产环境上存储数据的只推荐使用 MergeTree 系列的表引擎。
MergeTree 表引擎又分为单副本和多副本：

• 单副本的都是以 MergeTree 为名，例如 MergeTree 和 ReplacingMergeTree。

• 多副本的都加有前缀 Replicated，例如 ReplicatedMergeTree 和 ReplicatedReplacingMergeTree。

其中单副本的引擎不推荐在生产环境上使用，只可以在测试的时候使用。
因此，我们下方的使用案例，将会聚焦于多副本的引擎使用。
在这个例子中，我们要创建一个表名为 "test_table" 的表

4.1 创建本地数据表

CREATE TABLE test.test_table_local  ON CLUSTER  cluster_emr
(
    EventDate DateTime,
    CounterID UInt32,
    UserID UInt32
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{cluster}/{shard}/test/test_table_local', '{replica}')
PARTITION BY toYYYYMM(EventDate)
ORDER BY (CounterID, EventDate, intHash32(UserID));

语句说明：

1. 我们创建一个带有后缀名 _local 的表，这个表作为本地表，会真实存储 ClickHouse 的数据。

2. 使用 on cluster 句式分布式执行该 DDL。

3. 指定列信息，表引擎，分区字段，排序字段。

4. 这里面表引擎的内容，需要详细说一下：

   多副本表引擎的语法如下: ReplicatedMergeTree('zookeeper_name_configured_in_auxiliary_zookeepers:path', 'replica_name')。
   它只要求填写一个 ZK 的路径和 Replica 的名字，但是不关心其中具体的值，因此我们在 EMR 上需要对路径和名字做个使用规范：ZK 路径使用规范为：/clickhouse/tables/{cluster}/{shard}/test/test_table_local：

   其中 /clickhouse/tables/ 为固定前缀，防止路径冲突，{cluster}/{shard} 为上文提到的两个宏标志集群和分区信息，填写方式就是使用 {cluster}/{shard}，在建表语句中使用宏需要大括号；test/test_table_local 作为数据库和表名字，需要根据库表不同时，灵活修改替换。

   副本名字使用规范为：{replica}，使用 replica 宏即可。

   更多的表引擎使用方法，参考官方文档

4.2 创建分布式表

查询本地数据表，查询语句只会在当前节点执行，数据也只是其中一部分，若执行分布式查询，需要创建对应的分布式表。

CREATE TABLE IF NOT EXISTS test.test_table ON CLUSTER cluster_emr
AS test.test_table_local 
ENGINE = Distributed(cluster_emr, test, test_table_local, xxHash32(UserID));

语句说明：

1. 确定分布式表名，和分布式 DDL 语法

2. 指定分布式表字段信息，这里使用 AS 语法直接关联本地数据表，CK 会自动读取数据表的字段信息。

3. 指定 Engine 信息，分布式表的语法为：Distributed(cluster, database, table, sharding_key)

   • cluster：在这个例子中为 cluster_emr

   • database：这里为 test

   • table：为本地数据表的名字，即 test_table_local

   • sharding_key：表示数据如何分布，在这个样例中使用用户 ID 的 hash 值作为例子，用户也可以使用 rand() 函数，进行随机分区。

更多分布式表的使用方式，请参考官方文档

5 导入数据

通过 Insert 语句构造部分测试数据：

insert into test.test_table select '2020-02-02 00:00:00', number+1, number from numbers(100);

6 查询数据

6.1 查询行数

select count(*) from test.test_table;

6.2 点查询

select * from test.test_table where UserID=10;

6.3 聚合查询

select UserID, sum(CounterID)  from test.test_table group by UserID ;

更多查询语法，查看官方文档。