大数据技术之Clickhouse

一、概述

ClickHouse 是俄罗斯的Yandex于2016年开源的列式存储数据库（DBMS），使用C++语言编写，主要用于在线分析处理查询（OLAP），能够使用SQL查询实时生成分析数据报告。

1.1 适用场景

1. 绝大多数是读请求；

2. 数据以相当大的批次(> 1000行)更新，而不是单行更新;或者根本没有更新；

3. 已添加到数据库的数据不能修改；

4. 对于读取，从数据库中提取相当多的行，但只提取列的一小部分；

5. 宽表，即每个表包含着大量的列；

6. 查询相对较少(通常每台服务器每秒查询数百次或更少)；

7. 对于简单查询，允许延迟大约50毫秒；

8. 列中的数据相对较小：数字和短字符串(例如，每个URL 60个字节)；

9. 处理单个查询时需要高吞吐量(每台服务器每秒可达数十亿行)；

10. 事务不是必须的；

11. 对数据一致性要求低；

12. 每个查询有一个大表。除了他以外，其他的都很小；

13. 查询结果明显小于源数据。换句话说，数据经过过滤或聚合，因此结果适合于单个服务器的RAM中；

1.2 优缺点

1. 优点

跑分快（对于大多数查询而言，处理速度至少提高了100倍）；
向量化执行引擎：为了实现向量化执行，需要利用CPU的SIMD指令，即用单条指令操作多条数据；
可以实现流批一体；
支持shard + replication（类似HBase的Region）；
支持嵌套等复杂的数据结构；
使用MPP架构；（https://www.cnblogs.com/jianyungsun/p/9261632.html）

2. 缺点

不能完整支持事务（并发读写没有分离）；
不能很高吞吐量的修改和删除数据（行数据删除）；
由于索引的稀疏性，不适合基于Key来查询单个记录（不适合做Key-Value数据库），适合多维度的聚合操作；
对Hadoop生态不兼容（不适合用HDFS存储数据）；

1.3 特性

C++语言利用硬件资源优势
列式存储和压缩
单节点多核并行处理
索引（一级、二级、稀疏索引）
运用了大量算法
向量化引擎（使用了CPU的寄存器、SIMD指令-并行）
预计算模型
数据分片和分布查询

二、安装部署

2.1 环境准备

2.1.1 CentOS取消打开文件数限制

1.修改文件sudo vim /etc/security/limits.conf
sudo vim /etc/security/limits.d/90-nproc.conf2.在文件末尾添加：* soft nofile 65536
* hard nofile 65536
* soft nproc 131072
* hard nproc 131072

2.1.2 取消SELINUX

1.修改文件
sudo vim /etc/selinux/config2.修改内容
SELINUX=disabled

2.1.3 关闭防火墙

sudo systemctl disable firewalld
sudo systemctl stop firewalld

2.1.4 安装依赖

sudo yum install -y libtool
sudo yum install -y *unixODBC*

2.1.5 下载安装包

官网地址：https://clickhouse.tech/docs/zh/getting-started/install/

2.2 单机部署

2.2.1安装

执行安装命令（root用户）：

添加官方存储库
sudo yum install yum-utils
sudo rpm --import https://repo.clickhouse.tech/CLICKHOUSE-KEY.GPG
sudo yum-config-manager --add-repo https://repo.clickhouse.tech/rpm/stable/x86_64运行安装命令
sudo yum install clickhouse-server clickhouse-client

2.2.2 配置

修改配置文件/etc/clickhouse-server/config.xml

sudo vim /etc/clickhouse-server/config.xml

修改TCP端口，因为Clickhouse默认通信端口为9000，这与Hadoop中的HDFS端口冲突。

2. 把 <listen_host>::</listen_host> 的注解打开，这样的话才能让clickhouse被除本机以外的服务器访问。

2.2.4 启动

sudo systemctl start clickhouse-server
访问客户端：
clickhouse-client –-port 19000

2.3 集群部署

2.3.1 安装单节点Clickhouse

按照3.2单机部署的方式在每台机器（node1,node2,node3）进行安装。

2.3.2 分布式集群方案

2.3.2.1 MergeTree + Distributed

1. 建表方式

--1.创建本地表
CREATE TABLE db.tb (date Date, ……) ENGINE = MergeTree(date, (date, hour, datetime), 8192)--2.创建分布式表（相当于View，聚合本地表的结果）
CREATE TABLE db.tb_all (date Date, ……) ENGINE = Distributed(bip_ck_cluster, 'ck_cluster', 'test', rand())"

2. 架构图

MergeTree + Distributed的分布式架构方案，利用的是Distributed表的特性+MergeTree表的特性，分布式表不存储数据（View），数据来自本地表（Local表），将分布式表的数据分为3个shard，每台节点存储三分之一的数据，用户查询的时候是从分布式表所在的节点聚合从node1，node2，node3的查询结果，然后返回用户，写入数据可以写入分布式表（日增小于百万的数据可以使用，但是百万以上建议使用本地表），也可以分别将数据写入本地表（分散写入node1，node2，node3），当然你也可以只写入其中一台，那么使用方式就是单机版的。（https://blog.csdn.net/ashic/article/details/108725437）

1. 优势：架构简单，可以单机使用，可以分布式使用，关键在于表引擎的选择，并行的查询分布式表，性能非常棒。

2. 问题：

没有副本机制，数据丢失，查询结果会不一致；

对于查询节点的选择需要慎重的考虑，毕竟需要聚合所有查询节点的结果；

注：这里每一个Local表，即称为一个分片。

2.3.2.2 MergeTree + Distributed+副本

1. 建表方式

该方案的建表方式与方案一相同，只是本地表自身会带有副本。

--1.创建本地表
CREATE TABLE db.tb (date Date, ……) ENGINE = MergeTree(date, (date, hour, datetime), 8192)--2.分布式表（相当于View，聚合本地表的结果）
CREATE TABLE db.tb_all (date Date, ……) ENGINE = Distributed(bip_ck_cluster, 'ck_cluster', 'test', rand())"

注：需要修改配置文件，见3.3.2节。

2. 架构图

分布式架构2采用了架构1的特点和解决了架构1的问题，数据安全性得到了解决，结合CLickHouse集群的复制，有了副本，3个shard各自拥有三分之一的数据，每个shard有2个副本，数据一样。其中CK1的Shard有两副本，分别在Node01，Node02；Node02的shard也有两副本，分别在Node02，Node03；Node03的shard也是两副本，分别在Node01和Node03。

注：集群层面的副本。

1. 优势：数据的安全性有了保障，每一个shard有两个副本；数据的查询的并行度没有改变，但是因为副本的存在，shard节点数据的查询选择性多了。即使Node01挂了，不影响集群的查询服务。

2. 问题：

每一个shard需要同时负责分片和副本的数据写入工作，它很有可能称为写入的单点瓶颈。

2.3.2.3 ReplicatedMergeTree + Distributed

1. 建表方式

--1.创建本地表：
CREATE TABLE db.tb (date Date, ……) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{layer}-{shard}/table_name', '{replica}')--2.创建分布式表：
CREATE TABLE db.tb_all (date Date, ……) ENGINE = Distributed(bip_ck_cluster, 'test', 'test', rand())"

需要修改配置文件，见3.3.2节。

2. 架构图

ReplicatedMergeTree + Distributed的架构把MergeTree换成了ReplicatedMergeTree，本质上是将副本的数据同步策略，从基于Cluster的方式换成了基于复制表引擎+Zookeeper的方式，基于ReplicatedMergeTree + Distributed的架构方案，在查询并行度，数据的安全性，副本的安全性，数据的一致性上都考虑的比较好，也避免了社区提出的DistrDirMonitor的故障问题。

1. 优点：ReplicatedMergeTree里，共享同一个ZK路径的表，会相互同步数据，数据安全，查询性能不会有太大的问题。

注：副本的上限即为Clickhouse实例的个数。

2.3.2 新增配置文件

本次安装有三个节点，使用ReplicatedMergeTree + Distributed方案，配置为环形复制拓扑结构。

环形复制拓扑结构，虽然用比较少的节点，实现了分布式 + 副本的架构，但是同一个节点上会同时存在相同名称的两个表（Local表和副本表）。

这显然不起作用，因为Local表具有相同的表名，当它们位于同一服务器上时，ClickHouse无法区分一个Local/副本。这里的诀窍是把每个Local表放到一个单独的数据库中！ClickHouse允许为每个shard定义“default_database”，然后在查询时使用它，以便将特定表的查询路由到正确的数据库。

在/etc/clickhouse-server/config.d目录下创建一个名为metrika.xml，内容如下：

<yandex><clickhouse_remote_servers><!-- 集群名称--><ck_cluster><!--三个分片、每个分片两个副本--><shard><!--集群的第一个分片--><internal_replication>true</internal_replication><!--该分片的第一个副本--><replica><!--设置默认库--><default_database>shard_1</default_database><host>hadoop101</host><port>19000</port></replica><!--该分片的第二个副本--><replica><!--设置默认库--><default_database>shard_1</default_database><host>hadoop102</host><port>19000</port></replica></shard><shard><!--集群的第二个分片--><internal_replication>true</internal_replication><!--该分片的第一个副本--><replica><!--设置默认库--><default_database>shard_2</default_database><host>hadoop102</host><port>19000</port></replica><!--该分片的第二个副本--><replica><!--设置默认库--><default_database>shard_2</default_database><host>hadoop103</host><port>19000</port></replica></shard><shard><!--集群的第三个分片--><internal_replication>true</internal_replication><!--该分片的第一个副本--><replica><!--设置默认库--><default_database>shard_3</default_database><host>hadoop103</host><port>19000</port></replica><!--该分片的第二个副本--><replica><!--设置默认库--><default_database>shard_3</default_database><host>hadoop101</host><port>19000</port></replica></shard></ck_cluster></clickhouse_remote_servers><!--指定Zookeeper集群--><zookeeper-servers><node index="1"><host>hadoop101</host><port>2181</port></node><node index="2"><host>hadoop102</host><port>2181</port></node><node index="3"><host>hadoop103</host><port>2181</port></node></zookeeper-servers><!----><macros><!--不同机器放的分片数不一样--><shard>01</shard><!--不同机器放的副本数不一样--><replica>rep_01_1</replica></macros><!--不限制访问来源ip地址--><networks><ip>::/0</ip></networks><!--数据压缩方式，默认为lz4--><clickhouse_compression><case><!--可被压缩的数据块的最小大小，默认值10GB--><min_part_size>10000000000</min_part_size><!--可被压缩的数据块占全表大小的最小比例，默认值1%--><min_part_size_ratio>0.01</min_part_size_ratio><!--压缩算法，可选lz4和zstd--><method>lz4</method></case></clickhouse_compression>
</yandex>

2.3.4 修改配置文件

在配置文件/etc/clickhouse-server/config.xml中，新增

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

三、Clickhouse用法

3.1数据库引擎

延时引擎、Atomic引擎、MySQL引擎、数据库引擎。

3.1.1 延时引擎

在距最近一次访问间隔expiration_time_in_seconds时间段内，将表保存在内存中，仅适用于 *Log引擎表。

--创建数据库
CREATE DATABASE testlazy ENGINE = Lazy(expiration_time_in_seconds);

3.1.2 Atomic引擎

它是支撑无阻塞DROP和RENAME TABLE查询和原子EXCHANGE TABLES t1 AND t2查询。默认情况下使用原子数据库引擎。

--创建数据库
CREATE DATABASE test ENGINE = Atomic;

3.1.3 MySQL引擎

MySQL引擎用于将远程的MySQL服务器中的表映射到ClickHouse中，并允许您对表进行INSERT和SELECT查询，以方便您在ClickHouse与MySQL之间进行数据交换。

MySQL数据库引擎会将对其的查询转换为MySQL语法并发送到MySQL服务器中，因此您可以执行诸如SHOW TABLES或SHOW CREATE TABLE之类的操作。

CREATE DATABASE [IF NOT EXISTS] db_name [ON CLUSTER cluster]
ENGINE = MySQL('host:port', ['database' | database], 'user', 'password')

3.1.4 数据库引擎

默认情况下，ClickHouse使用自己的数据库引擎，该引擎提供可配置的表引擎和所有支持的SQL语法.

3.2 表引擎

表引擎是Clickhouse的一大特色。可以说，表引擎决定了如何存储标的数据。包括：

1）数据的存储方式和位置，写到哪里以及从哪里读取数据；

2）支持哪些查询以及如何支持；

3）并发数据访问；

4）索引的使用（如果存在）；

5）是否可以执行多线程请求；

6）数据复制参数；

表引擎的使用方式就是必须显形在创建表时定义该表使用的引擎，以及引擎使用的相关参数。如：

create table t_tinylog ( id String, name String) Engine=TinyLog;

3.2.1 MergeTree

Clickhouse 中最强大的表引擎当属 MergeTree （合并树）引擎及该系列（*MergeTree）中的其他引擎。

MergeTree 系列的引擎被设计用于插入极大量的数据到一张表当中。数据可以以数据片段的形式一个接着一个的快速写入，数据片段在后台按照一定的规则进行合并。相比在插入时不断修改（重写）已存储的数据，这种策略会高效很多。

1. 特点

存储的数据按主键排序：这使得你能够创建一个小型的稀疏索引来加快数据检索。
支持数据分区：在相同数据集和相同结果集的情况下 ClickHouse 中某些带分区的操作会比普通操作更快。查询中指定了分区键时 ClickHouse 会自动截取分区数据。这也有效增加了查询性能。
支持数据副本：ReplicatedMergeTree 系列的表提供了数据副本功能。更多信息见3.2.5节。
支持数据采样：需要的话，你可以给表设置一个采样方法。

2. 建表

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1] [TTL expr1],name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2] [TTL expr2],...INDEX index_name1 expr1 TYPE type1(...) GRANULARITY value1,INDEX index_name2 expr2 TYPE type2(...) GRANULARITY value2
) ENGINE = MergeTree() //指定表引擎
ORDER BY expr
[PARTITION BY expr]
[PRIMARY KEY expr]
[SAMPLE BY expr]
[TTL expr [DELETE|TO DISK 'xxx'|TO VOLUME 'xxx'], ...]
[SETTINGS name=value, ...]

注意

order by是MergeTree中唯一一个必填项，甚至比primary key 还重要，因为当用户不设置主键的情况，很多处理会依照order by的字段进行处理（去重和汇总）。

要求：主键必须是order by字段的前缀字段。

3.2.2 LogFamily

主要包括：StripeLog、Log、TinyLog，这些引擎是为了需要写入许多小数据量（少于一百万行）的表的场景而开发的。

3.2.3 Integrations

常用的集成引擎：JDBC、MySQL、HDFS、Kafka。

3.2.4 Special

Memory：内存引擎，数据以未压缩的原始形式直接保存在内存当中，服务器重启数据就会消失。读写操作不会相互阻塞，不支持索引。简单查询下有非常非常高的性能表现（超过10G/s）。

一般用到它的地方不多，除了用来测试，就是在需要非常高的性能，同时数据量又不太大（上限大概 1 亿行）的场景。

3.2.5 数据副本

只有MergeTree系列引擎支持副本。副本是表级别的，不是整个服务器级的（不同与分布式方式二）。所以，服务器里可以同时有复制表和非复制表。

副本不依赖分片。每个分片有它自己的独立副本。支持副本的MergeTree如下：

ReplicatedMergeTree
ReplicatedSummingMergeTree
ReplicatedReplacingMergeTree
ReplicatedAggregatingMergeTree
ReplicatedCollapsingMergeTree
ReplicatedVersionedCollapsingMergetree
ReplicatedGraphiteMergeTree

副本的创建依赖于Zookeeper集群，如果配置文件中没有设置 ZooKeeper ，则无法创建复制表，并且任何现有的复制表都将变为只读。ReplicatedMergeTree引擎族在ZK中存储大量数据，包括且不限于表结构信息、元数据、操作日志、副本状态、数据块校验值、数据part merge过程中的选主信息等等。可见，ZK在复制表机制下扮演了元数据存储、日志框架、分布式协调服务三重角色，任务很重，所以需要额外保证ZK集群的可用性以及资源（尤其是硬盘资源）。

1. Replicated*MergeTree 参数

zoo_path：ZooKeeper 中该表的路径。
replica_name：ZooKeeper 中的该表的副本名称。

2. 建表语句

CREATE TABLE table_name
(EventDate DateTime,CounterID UInt32,UserID UInt32
) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{layer}-{shard}/table_name', '{replica}')
PARTITION BY toYYYYMM(EventDate)
ORDER BY (CounterID, EventDate, intHash32(UserID))
SAMPLE BY intHash32(UserID)

3.2.6 Distributed

Clickhouse是支持分布式存储、分布式计算的概念的，虽然上节保证了副本机制，但是还没有介绍到分布式集群的内容。Clickhouse的分布式集群的基础是“分片”的概念，分片即2.3.2.1节中提到的Local表，是通过部署到各个节点的Local表，从而实现分布式存储和分布式计算。

通过分片把一份完整的数据进行切分，不同的分片分布到不同的节点上，再通过Distributed表引擎把数据拼接起来一同使用。Distributed表引擎本身不存储数据，有点类似于MyCat之于MySql，成为一种中间件，通过分布式逻辑表来写入、分发、路由来操作多台节点不同分片的分布式数据。

1. 参数

cluster_name：服务器配置文件中的集群名，在/etc/metrika.xml中配置的。具体配置见前文。
database：数据库名。
table：表名。
sharding_key：数据分片键。

2. 建表语句

CREATE TABLE table_name
(EventDate DateTime,CounterID UInt32,UserID UInt32
) ENGINE = Distributed(cluster_name, database, table, [sharding_key])
PARTITION BY toYYYYMM(EventDate)
ORDER BY (CounterID, EventDate, intHash32(UserID))
SAMPLE BY intHash32(UserID)

3.3 SQL语法

严格区分大小写。

详见官网：https://clickhouse.tech/docs/zh/sql-reference/syntax/