目录
- 第一章 clickhouse概述
- 第二章 单机版安装
- 第三章 集群安装
- 第四章 入门操作
- 第五章 库表引擎
- 第六章 数据类型
- 第七章 函数
- 第八章 bitmap
第一章 clickhouse概述
中文官网:https://clickhouse.tech/docs/zh/
阿里云clickhouse官网:https://www.aliyun.com/product/clickhouse (不是特别建议,可以补充看)
1.1 clickhouse的定义
ClickHouse是一个开源的用于联机分析(OLAP)的列式数据库管理系统(DBMS)。
- 大多数是读请求
- 数据总是以相当大的批(> 1000 rows)进行写入
- 不修改已添加的数据
- 每次查询都从数据库中读取大量的行,但是同时又仅需要少量的列
- 宽表,即每个表包含着大量的列
- 较少的查询(通常每台服务器每秒数百个查询或更少)
- 对于简单查询,允许延迟大约50毫秒
- 列中的数据相对较小: 数字和短字符串(例如,每个URL 60个字节)
- 处理单个查询时需要高吞吐量(每个服务器每秒高达数十亿行)
- 事务不是必须的
- 对数据一致性要求低
- 每一个查询除了一个大表外都很小
- 查询结果明显小于源数据,换句话说,数据被过滤或聚合后能够被盛放在单台服务器的内存中
很容易可以看出,OLAP场景与其他通常业务场景(例如,OLTP或K/V)有很大的不同, 因此想要使用OLTP或Key-Value数据库去高效的处理分析查询场景,并不是非常完美的适用方案。例如,使用OLAP数据库去处理分析请求通常要优于使用MongoDB或Redis去处理分析请求。
1.2 clickhouse的来源
ClickHouse最初是一款名为Yandex.Metrica的产品,主要用于WEB流量分析。ClickHouse的全称是Click Stream,Data WareHouse,简称ClickHouse。也是战斗名族俄罗斯([Russia)的第二款强大开源的高性能产品之一。
1.3 clickhouse应用领域
ClickHouse非常适用于商业智能领域,除此之外,它也能够被广泛应用于广告流量、Web、App流量、电信、金融、电子商务、信息安全、网络游戏、物联网等众多其他领域。
1.4 clickhouse优点
https://clickhouse.tech/docs/zh/introduction/distinctive-features/
- 支持完备的SQL操作
- 真正的列式存储与数据压缩
- 向量化(列的一部分)执行引擎
- 关系型模型(与传统数据库类似)
丰富的表引擎
- 并行处理
- 在线查询
- 数据分片
1.5 clickhouse缺点
- 不支持事务。为啥不支持事务?因面向列
- 不擅长根据主键按行粒度进行查询(虽然支持),故不应该把ClickHouse当作Key-Value数据库使用。
- 不擅长按行删除数据(虽然支持)
clickhouse和hbase的区别?
第二章 单机版安装
2.1 rpm安装(也可以在线yum安装)
1 rpm下载地址
https://repo.yandex.ru/clickhouse/rpm/stable/x86_64
2 下载对应版本
[root@hadoop01 home]# ll /home/clickhouse/clickhouse-*
/home/clickhouse/clickhouse-client-20.5.2.7-2.noarch.rpm
/home/clickhouse/clickhouse-common-static-20.5.2.7-2.x86_64.rpm
/home/clickhouse/clickhouse-server-20.5.2.7-2.noarch.rpm
/home/clickhouse/clickhouse-test-20.5.2.7-2.noarch.rpm
3 安装已下载的版本(有顺序关系或者yum -y install /home/clickhouse-*.rpm)
#执行如下命令(不用考虑顺序)
[root@hadoop01 home]# yum -y install /home/clickhouse/clickhouse-*.rpm
#单个安装,需要考虑依赖顺序
[root@hadoop01 home]# yum -y install /home/clickhouse/clickhouse-common-static-20.5.2.7-2.x86_64.rpm
[root@hadoop01 home]# yum -y install /home/clickhouse/clickhouse-client-20.5.2.7-2.noarch.rpm
[root@hadoop01 home]# yum -y install /home/clickhouse/clickhouse-server-20.5.2.7-2.noarch.rpm
[root@hadoop01 home]# yum -y install /home/clickhouse/clickhouse-test-20.5.2.7-2.noarch.rpm
问题1:
--> Finished Dependency Resolution
Error: Package: clickhouse-test-20.5.2.7-2.noarch (/clickhouse-test-20.5.2.7-2.noarch)
Requires: perl(JSON::XS)
You could try using --skip-broken to work around the problem
** Found 2 pre-existing rpmdb problem(s), 'yum check' output follows:
2:postfix-2.10.1-7.el7.x86_64 has missing requires of libmysqlclient.so.18()(64bit)
2:postfix-2.10.1-7.el7.x86_64 has missing requires of libmysqlclient.so.18(libmysqlclient_18)(64bit)
解决办法:
缺少perl(JSON::XS)这个包
# 安装依赖
[root@hadoop01 home]# yum install -y epel-release perl-JSON-XS
yum安装
#准备YUM源
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/clickhouse.repo
#安装server&client
sudo yum install clickhouse-server clickhouse-client
#启动server。(9000端口用于接受客户端请求;9009用于集群复制数据;8123是http请求端口)
sudo /etc/init.d/clickhouse-server start
systemctl start clickhouse-server
#启动client
clickhouse-client --host hadoop01
2.2 配置
配置config.xml
[root@hadoop01 home]# mv /etc/clickhouse-server/config.xml /etc/clickhouse-server/config.xml_bak
[root@hadoop01 home]# vi /etc/clickhouse-server/config.xml
#覆盖内容如下
<?xml version="1.0"?>
<yandex>
<!-- 配置日志文件 -->
<logger>
<level>trace</level>
<log>/data/clickhouse/logs/server.log</log>
<errorlog>/data/clickhouse/logs/error.log</errorlog>
<size>1000M</size>
<count>10</count>
</logger>
<!--配置http、tcp端口和监听地址-->
<http_port>8123</http_port>
<tcp_port>9000</tcp_port><!--默认9000-->
<interserver_http_port>9009</interserver_http_port><!--默认9009-->
<listen_host>::</listen_host>
<path>/data/clickhouse/clickhousedata/</path>
<tmp_path>/data/clickhouse/tmp/</tmp_path>
<users_config>users.xml</users_config><!--用户名default 密码为空-->
<default_profile>default</default_profile>
<default_database>default</default_database>
<remote_servers incl="clickhouse_remote_servers" />
<zookeeper incl="zookeeper-servers" optional="true" />
<macros incl="macros" optional="true" />
<!--默认/etc/metrika.xml-->
<include_from>/etc/clickhouse-server/metrika.xml</include_from>
<mark_cache_size>5368709120</mark_cache_size>
</yandex>
支持高可用创建 /etc/clickhouse-server/metrika.xml
[root@hadoop01 home]# vi /etc/clickhouse-server/metrika.xml
#覆盖如下内容即可
<yandex>
<clickhouse_remote_servers>
<news_ck_cluster><!--自定义集群名-->
<shard><!--一个分片-->
<weight>1</weight><!-- 写入数据的权重 -->
<internal_replication>false</internal_replication><!-- 复制数据是否写入全部副本-->
<replica>
<host>192.168.216.111</host>
<port>9000</port>
<user>default</user>
<password>123456</password>
</replica>
</shard>
</news_ck_cluster>
</clickhouse_remote_servers>
<!--宏配置,主要用于创建分布式复制表时替换参数,macros若省略,则建复制表时每个分片需指定zookeeper路径及副本名称-->
<macros>
<replica>192.168.216.111</replica> <!--默认当前服务器ip即可-->
</macros>
<networks>
<ip>::/0</ip>
</networks>
<zookeeper-servers>
<node index="1">
<host>192.168.216.111</host>
<port>2181</port>
</node>
</zookeeper-servers>
<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>
**修改/etc/clickhouse-server/users.xml **
[root@hadoop01 home]# mv /etc/clickhouse-server/users.xml /etc/clickhouse-server/users.xml_bak
[root@hadoop01 home]# vi /etc/clickhouse-server/users.xml
#覆盖如下内容
<?xml version="1.0"?>
<yandex>
<profiles>
<!-- 读写用户设置 -->
<default>
<max_memory_usage>10000000000</max_memory_usage>
<use_uncompressed_cache>0</use_uncompressed_cache>
<load_balancing>random</load_balancing>
</default>
<!-- 只读用户设置 -->
<readonly>
<max_memory_usage>10000000000</max_memory_usage>
<use_uncompressed_cache>0</use_uncompressed_cache>
<load_balancing>random</load_balancing>
<readonly>1</readonly>
</readonly>
</profiles>
<!-- 配额 -->
<quotas>
<!-- Name of quota. -->
<default>
<interval>
<duration>3600</duration>
<queries>0</queries>
<errors>0</errors>
<result_rows>0</result_rows>
<read_rows>0</read_rows>
<execution_time>0</execution_time>
</interval>
</default>
</quotas>
<users>
<!-- 读写用户 -->
<default>
<password_sha256_hex>8d969eef6ecad3c29a3a629280e686cf0c3f5d5a86aff3ca12020c923adc6c92</password_sha256_hex>
<networks incl="networks" replace="replace">
<ip>::/0</ip>
</networks>
<profile>default</profile>
<quota>default</quota>
</default>
<!-- 只读用户 -->
<readonly>
<password_sha256_hex>8d969eef6ecad3c29a3a629280e686cf0c3f5d5a86aff3ca12020c923adc6c92</password_sha256_hex>
<networks incl="networks" replace="replace">
<ip>::/0</ip>
</networks>
<profile>readonly</profile>
<quota>default</quota>
</readonly>
</users>
</yandex>
2.3 服务启动
后台服务启动
#推荐使用启动命令:
[root@hadoop01 home]# clickhouse-server --config-file=/etc/clickhouse-server/config.xml
# centos6、7命令
[root@hadoop01 home]# service clickhouse-server status
#或者 centos7.x命令
[root@hadoop01 home]# systemctl status/start/stop clickhouse-server
2.4 客户端连接
[root@hadoop01 clickhouse-server]# clickhouse-client \
--host=localhost \
--port=9000 \
--user=default \
--password=123456
hadoop01 :) show databases;
hadoop01 :) show tables;
hadoop01 :) use default;
hadoop01 :) select * from system.clusters;
SELECT *
FROM system.clusters
┌─cluster─────────┬─shard_num─┬─shard_weight─┬─replica_num─┬─host_name───────┬─host_address────┬─port─┬─is_local─┬─user────┬─default_database─┬─errors_count─┬─estimated_recovery_time─┐
│ news_ck_cluster │ 1 │ 1 │ 1 │ 192.168.216.111 │ 192.168.216.111 │ 9000 │ 0 │ default │ │ 0 │ 0 │
│ news_ck_cluster │ 1 │ 1 │ 2 │ 192.168.216.112 │ 192.168.216.112 │ 9000 │ 0 │ default │ │ 0 │ 0 │
│ news_ck_cluster │ 2 │ 1 │ 1 │ 192.168.216.112 │ 192.168.216.112 │ 9000 │ 1 │ default │ │ 0 │ 0 │
│ news_ck_cluster │ 2 │ 1 │ 2 │ 192.168.216.113 │ 192.168.216.113 │ 9000 │ 0 │ default │ │ 0 │ 0 │
│ news_ck_cluster │ 3 │ 1 │ 1 │ 192.168.216.111 │ 192.168.216.111 │ 9000 │ 0 │ default │ │ 0 │ 0 │
│ news_ck_cluster │ 3 │ 1 │ 2 │ 192.168.216.113 │ 192.168.216.113 │ 9000 │ 0 │ default │ │ 0 │ 0 │
└─────────────────┴───────────┴──────────────┴─────────────┴─────────────────┴─────────────────┴──────┴──────────┴─────────┴──────────────────┴──────────────┴─────────────────────────┘
6 rows in set. Elapsed: 0.010 sec.
目录说明
目录:/var/log/clickhouse-server 服务日志目录
目录:/etc/clickhouse-server 包含clickhouse的两个配置文件,config.xml和user.xml
脚本:/etc/init.d/clickhouse-server 手动启动clickhouse服务的脚本
目录:/var/lib/clickhouse/data clickhouse数据的存储目录,以数据库和表名进行区分
目录:/etc/metrika.xml 默认没有,读取集群配置
第三章 集群安装
3.1 安装
其它服务器和单机版一样,执行clickhouse的安装操作。
[root@hadoop01 home]# scp -r /home/clickhouse/ hadoop02:/home/
[root@hadoop01 home]# scp -r /home/clickhouse/ hadoop03:/home/
# 安装依赖
[root@hadoop02 home]# yum install -y epel-release
[root@hadoop02 home]# yum install -y perl-JSON-XS
[root@hadoop03 home]# yum install -y epel-release
[root@hadoop03 home]# yum install -y perl-JSON-XS
# 安装clickhouse
[root@hadoop02 home]# yum -y install /home/clickhouse/clickhouse-*.rpm
[root@hadoop03 home]# yum -y install /home/clickhouse/clickhouse-*.rpm
3.2 配置
创建 /etc/clickhouse-server/metrika.xml
#yml安装下,只添加metrika.xml即可
[root@hadoop01 home]# vi /etc/clickhouse-server/metrika.xml
#覆盖如下内容即可
<yandex>
<clickhouse_remote_servers>
<news_ck_cluster>
<shard>
<weight>1</weight>
<internal_replication>false</internal_replication>
<replica>
<host>192.168.216.111</host>
<port>9000</port>
<user>default</user>
<password>123456</password>
</replica>
<replica>
<host>192.168.216.112</host>
<port>9000</port>
<user>default</user>
<password>123456</password>
</replica>
</shard>
<shard>
<weight>1</weight>
<internal_replication>false</internal_replication>
<replica>
<host>192.168.216.112</host>
<port>9000</port>
<user>default</user>
<password>123456</password>
</replica>
<replica>
<host>192.168.216.113</host>
<port>9000</port>
<user>default</user>
<password>123456</password>
</replica>
</shard>
<shard>
<weight>1</weight>
<internal_replication>false</internal_replication>
<replica>
<host>192.168.216.111</host>
<port>9000</port>
<user>default</user>
<password>123456</password>
</replica>
<replica>
<host>192.168.216.113</host>
<port>9000</port>
<user>default</user>
<password>123456</password>
</replica>
</shard>
</news_ck_cluster>
</clickhouse_remote_servers>
<macros>
<replica>192.168.216.111</replica> <!--默认当前服务器ip即可-->
</macros>
<networks>
<ip>::/0</ip>
</networks>
<zookeeper-servers>
<node index="1">
<host>192.168.216.111</host>
<port>2181</port>
</node>
<node index="2">
<host>192.168.216.112</host>
<port>2181</port>
</node>
<node index="3">
<host>192.168.216.113</host>
<port>2181</port>
</node>
</zookeeper-servers>
<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>
复制config.xml users.xml metrika.xml到其它服务器
# copy配置文件
[root@hadoop01 clickhouse-server]# scp -r config.xml users.xml metrika.xml hadoop02:/etc/clickhouse-server
[root@hadoop01 clickhouse-server]# scp -r config.xml users.xml metrika.xml hadoop03:/etc/clickhouse-server
3.3 服务启动
推荐使用启动命令:
[root@hadoop01 home]# clickhouse-server --config-file=/etc/clickhouse-server/config.xml
[root@hadoop02 home]# clickhouse-server --config-file=/etc/clickhouse-server/config.xml
[root@hadoop03 home]# clickhouse-server --config-file=/etc/clickhouse-server/config.xml
3.4 客户端连接
[root@hadoop01 clickhouse-server]# clickhouse-client \
--host=localhost \
--port=9000 \
--user=default \
--password=123456
注:
集群模式需要依赖zk来存储元数据,所以需要先启动zk,才能启动clickhouse。
第四章 入门操作
4.1 单机基础操作
4.1.1 创建数据库
- 语法
CREATE DATABASE [IF NOT EXISTS] db_name [ON CLUSTER cluster] [ENGINE = engine(...)]
- 例子
CREATE DATABASE IF NOT EXISTS chtest; --使用默认库引擎创建库
默认情况下,ClickHouse使用的是原生的数据库引擎Ordinary(在此数据库下可以使用任意类型的表引擎,在绝大多数情况下都只需使用默认的数据库引擎)。当然也可以使用Lazy引擎和MySQL引擎,比如使用MySQL引擎,可以直接在ClickHouse中操作MySQL对应数据库中的表。假设MySQL中存在一个名为clickhouse的数据库,可以使用下面的方式连接MySQL数据库。
-- 语法
CREATE DATABASE [IF NOT EXISTS] db_name [ON CLUSTER cluster]
ENGINE = MySQL('host:port', ['database' | database], 'user', 'password')
--操作
hadoop01 :) CREATE DATABASE mysql_db ENGINE = MySQL('192.168.216.111:3306', 'sales_source', 'root', 'root');
hadoop01 :) use mysql_db;
hadoop01 :) show tables;
SHOW TABLES
┌─name────────┐
│ customer │
│ product │
│ sales_order │
└─────────────┘
hadoop01 :) select * from product limit 1;
SELECT *
FROM product
LIMIT 1
┌─product_code─┬─product_name─┬─product_category─┐
│ 1 │ Hard Disk │ Storage │
└──────────────┴──────────────┴──────────────────┘
注:
mysql数据库引擎的库下是不支持创建表。
4.1.2 创建表
- 语法
CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]( name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1] [compression_codec] [TTL expr1], name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2] [compression_codec] [TTL expr2], ...) ENGINE = engine
- 示例
-- 注意首字母大写
-- 建表
hadoop01 :) use default;
hadoop01 :) create table test(id Int32,name String) engine=Memory; --内存引擎表
hadoop01 :) insert into test(id,name) values(110,'goudan'); --字符串不能使用双引号
hadoop01 :) select * from test;
SELECT *
FROM test
┌──id─┬─name───┐
│ 110 │ goudan │
└─────┴────────┘
上面命令创建了一张内存表,即使用Memory引擎。表引擎决定了数据表的特性,也决定了数据将会被如何存储及加载。Memory引擎是ClickHouse最简单的表引擎,数据只会被保存在内存中,在服务重启时数据会丢失
。一般应用于中间表或者临时表。
4.2 分布式DDL操作
默认情况下,CREATE、DROP、ALTER、RENAME操作仅仅在当前执行该命令的server上生效。
在集群环境下,可以使用ON CLUSTER语句,这样就可以在整个集群发挥作用,每个节点同时创建表。
比如创建一张分布式表:
hadoop01 :) CREATE TABLE IF NOT EXISTS user_cluster ON CLUSTER news_ck_cluster(
id Int32,
name String) ENGINE = Distributed(news_ck_cluster, default, user_local,id);
Distributed表引擎的定义形式如下所示:关于ClickHouse的表引擎,后续章节做详细解释。
Distributed(cluster_name, database_name, local_table_name[, sharding_key])
各个参数的含义分别如下:
- cluster_name:集群名称,与集群配置中的自定义名称相对应。
- database_name:数据库名称
- table_name:表名称,映射到每台服务器中的表名称。
- sharding_key:可选的,用于分片的key值,在数据写入的过程中,分布式表会依据分片key的规则,将数据分布到各个节点的本地表。
创建完成上面的分布式表时,在每台机器上查看表,发现每台机器上都存在一张刚刚创建好的表。
接下来就需要创建本地表了,在每台机器上分别创建一张本地表:
-- 没有自动创建可以手动创建
hadoop01 :) CREATE TABLE IF NOT EXISTS user_local
(
id Int32,
name String
)ENGINE = MergeTree()
ORDER BY id
PARTITION BY id
PRIMARY KEY id;
我们先在一台机器上,对user_local表进行插入数据,然后再查询user_cluster表:
-- 插入数据
hadoop01 :) INSERT INTO user_local VALUES(1,'tom'),(2,'jack');
-- 查询user_cluster表,可见通过user_cluster表可以操作所有的user_local表
hadoop01 :) select * from user_cluster;
┌─id─┬─name─┐
│ 2 │ jack │
└────┴──────┘
┌─id─┬─name─┐
│ 1 │ tom │
└────┴──────┘
接下来,我们再向user_cluster中插入一些数据,观察user_local表数据变化,可以发现数据被分散存储到了其他节点上了。
-- 向user_cluster插入数据
hadoop01 :) INSERT INTO user_cluster VALUES(3,'lilei'),(4,'lihua');
-- 查看user_cluster数据
hadoop01 :) select * from user_cluster;
┌─id─┬─name─┐
│ 2 │ jack │
└────┴──────┘
┌─id─┬─name──┐
│ 3 │ lilei │
└────┴───────┘
┌─id─┬─name─┐
│ 1 │ tom │
└────┴──────┘
┌─id─┬─name──┐
│ 4 │ lihua │
└────┴───────┘
-- 在hadoop01上查看user_local
hadoop01 :) select * from user_local;
┌─id─┬─name─┐
│ 2 │ jack │
└────┴──────┘
┌─id─┬─name──┐
│ 3 │ lilei │
└────┴───────┘
┌─id─┬─name─┐
│ 1 │ tom │
└────┴──────┘
-- 在hadoop02上查看user_local
hadoop02 :) select * from user_local;
┌─id─┬─name──┐
│ 4 │ lihua │
└────┴───────┘
注:
1、创建分布式表是读时检查的机制,也就是说对创建分布式表和本地表的顺序并没有强制要求。
2、在上面的语句中使用了ON CLUSTER分布式DDL,这意味着在集群的每个分片节点上,都会创建一张Distributed表,这样便可以从其中任意一端发起对所有分片的读、写请求。
3、分布式表映射到每台服务器一张本地表。
4.3 为什么clickhouse快
列式数据库更适合于OLAP场景(对于大多数查询而言,处理速度至少提高了100倍)
输入/输出
- 针对分析类查询,通常只需要读取表的一
小部分列
。在列式数据库中你可以只读取你需要的数据。例如,如果只需要读取100列中的5列,这将帮助你最少减少20倍的I/O消耗。 - 由于数据总是打包成批量读取的,所以压缩是非常容易的。同时数据按列分别存储这也更
容易压缩
。这进一步降低了I/O的体积。 - 由于I/O的降低,这将帮助更多的数据被系统缓存。
Row-oriented DBMS
Column-oriented DBMS
性能
-
单个大查询的吞吐量
吞吐量可以使用每秒处理的行数或每秒处理的字节数来衡量。如果数据被放置在page cache中,则一个不太复杂的查询在单个服务器上大约能够以
2-10GB/s
(未压缩)的速度进行处理(对于简单的查询,速度可以达到30GB/s)。如果数据没有在page cache中的话,那么速度将取决于你的磁盘系统和数据的压缩率。例如,如果一个磁盘允许以400MB/s的速度读取数据,并且数据压缩率是3,则数据的处理速度为1.2GB/s。这意味着,如果你是在提取一个10字节的列,那么它的处理速度大约是1-2亿行每秒。对于分布式处理,处理速度几乎是线性扩展的,但这受限于聚合或排序的结果不是那么大的情况下。
-
处理短查询的延迟时间
如果一个查询使用主键并且没有太多行(几十万)进行处理,并且没有查询太多的列,那么在数据被page cache缓存的情况下,它的延迟应该小于50毫秒(
在最佳的情况下应该小于10毫秒)。 否则,延迟取决于数据的查找次数。如果你当前使用的是HDD,在数据没有加载的情况下,查询所需要的延迟可以通过以下公式计算得知: 查找时间(10 ms) * 查询的列的数量 * 查询的数据块的数量。 -
处理大量短查询的吞吐量
在相同的情况下,ClickHouse可以在单个服务器上每秒处理数百个查询
(在最佳的情况下最多可以处理数千个)。但是由于这不适用于分析型场景。因此我们建议每秒最多查询100次。 -
数据的写入性能
我们建议每次写入不少于1000行的批量写入,或每秒不超过一个写入请求
。当使用tab-separated格式将一份数据写入到MergeTree表中时,写入速度大约为50到200MB/s。如果您写入的数据每行为1Kb,那么写入的速度为50,000到200,000行每秒。如果您的行更小,那么写入速度将更高。为了提高写入性能,您可以使用多个INSERT进行并行写入,这将带来线性的性能提升。
第五章 库表引擎
官网地址:https://clickhouse.tech/docs/en/engines/table-engines/
库表引擎分类如下图:
表引擎(table engine)。类似mysql中的InnoDB和MyISAM存储引擎。不同的存储引擎提供不同的存储机制、索引方式、锁定水平等功能
,也可以称之为表类型。ClickHouse提供了丰富的表引擎,这些不同的表引擎也代表着不同的表类型。比如数据表拥有何种特性、数据以何种形式被存储以及如何被加载。本章会对ClickHouse中常见的表引擎进行介绍,主要包括以下内容:
- 表引擎的作用是什么
- MergeTree系列引擎
- Log家族系列引擎
- 外部集成表引擎
- 其他特殊的表引擎
5.1 表引擎作用
- 决定表存储在哪里以及以何种方式存储
- 支持哪些查询以及如何支持
- 并发数据访问
- 索引的使用
- 是否可以执行多线程请求
- 数据复制参数
5.2 表引擎分类
本章第一张图说明过。
引擎分类 | 引擎名称 |
---|---|
MergeTree系列 | MergeTree 、ReplacingMergeTree 、SummingMergeTree 、 AggregatingMergeTree CollapsingMergeTree 、 VersionedCollapsingMergeTree 、GraphiteMergeTree |
Log系列 | TinyLog 、StripeLog 、Log |
Integration Engines | Kafka 、MySQL、ODBC 、JDBC、HDFS |
Special Engines | Distributed 、MaterializedView、 Dictionary 、Merge 、File、Null 、Set 、Join 、 URL View、Memory 、 Buffer |
5.3 Log系列表引擎
5.3.1 应用场景
Log系列表引擎功能相对简单,主要用于快速写入小表(1百万行左右的表),然后全部读出的场景。即一次写入多次查询。
5.3.2 Log系列表引擎的特点
共性特点
- 数据存储在磁盘上
- 当写数据时,将数据追加到文件的末尾
- 不支持并发读写,当向表中写入数据时,针对这张表的查询会被阻塞,直至写入动作结束
- 不支持索引
- 不支持原子写:如果某些操作(异常的服务器关闭)中断了写操作,则可能会获得带有损坏数据的表
- 不支持ALTER操作(这些操作会修改表设置或数据,比如delete、update等等)
区别
-
TinyLog
TinyLog是Log系列引擎中功能简单、性能较低的引擎。它的存储结构由数据文件和元数据两部分组成。其中,数据文件是按列独立存储的,也就是说每一个列字段都对应一个文件。除此之外,TinyLog不支持并发数据读取。
-
StripLog支持并发读取数据文件,当读取数据时,ClickHouse会使用多线程进行读取,每个线程处理一个单独的数据块。另外,StripLog将所有列数据存储在同一个文件中,减少了文件的使用数量。
-
Log支持并发读取数据文件,当读取数据时,ClickHouse会使用多线程进行读取,每个线程处理一个单独的数据块。Log引擎会将每个列数据单独存储在一个独立文件中。
5.3.3 TinyLog表引擎使用
该引擎适用于一次写入,多次读取的场景。对于处理小批数据的中间表可以使用该引擎。值得注意的是,使用大量的小表存储数据,性能会很低。
CREATE TABLE emp_tinylog
(
emp_id UInt16 COMMENT '员工id',
name String COMMENT '员工姓名',
work_place String COMMENT '工作地点',
age UInt8 COMMENT '员工年龄',
depart String COMMENT '部门',
salary Decimal32(2) COMMENT '工资'
) ENGINE = TinyLog();
INSERT INTO emp_tinylog
VALUES (1, 'tom', '上海', 25, '技术部', 20000),
(2, 'jack', '上海', 26, '人事部', 10000);
INSERT INTO emp_tinylog
VALUES (3, 'bob', '北京', 33, '财务部', 50000),
(4, 'tony', '杭州', 28, '销售事部', 50000);
进入默认数据存储目录,查看底层数据存储形式,可以看出:TinyLog引擎表每一列都对应的文件:
[root@hadoop01 ~]# ll /data/clickhouse/clickhousedata/data/default/emp_tinylog/
total 28
-rw-r----- 1 root root 56 Sep 24 00:59 age.bin
-rw-r----- 1 root root 97 Sep 24 00:59 depart.bin
-rw-r----- 1 root root 60 Sep 24 00:59 emp_id.bin
-rw-r----- 1 root root 70 Sep 24 00:59 name.bin
-rw-r----- 1 root root 68 Sep 24 00:59 salary.bin
-rw-r----- 1 root root 185 Sep 24 00:59 sizes.json
-rw-r----- 1 root root 80 Sep 24 00:59 work_place.bin
## 在sizes.json文件内使用JSON格式记录了每个.bin文件内对应的数据大小的信息
## 查看sizes.json数据
[root@hadoop01 ~]# cat /data/clickhouse/clickhousedata/data/default/emp_tinylog/sizes.json
{"yandex":{"age%2Ebin":{"size":"56"},"depart%2Ebin":{"size":"97"},"emp_id%2Ebin":{"size":"60"},"name%2Ebin":{"size":"70"},"salary%2Ebin":{"size":"68"},"work_place%2Ebin":{"size":"80"}}}
当我们执行ALTER操作时会报错,说明该表引擎不支持ALTER操作(ALTER仅支持MergeTree族表引擎)
hadoop01 :) ALTER TABLE emp_tinylog DELETE WHERE emp_id = 5;
ALTER TABLE emp_tinylog
DELETE WHERE emp_id = 5
Received exception from server (version 20.5.2):
Code: 48. DB::Exception: Received from localhost:9000. DB::Exception: Mutations are not supported by storage TinyLog.
5.3.4 StripLog表引擎
相比TinyLog而言,StripeLog拥有更高的查询性能
(拥有.mrk标记文件,支持并行查询),同时其使用了更少的文件描述符(所有数据使用同一个文件保存)。
--创建StripeLog表
CREATE TABLE emp_stripelog (emp_id UInt16 COMMENT '员工id',name String COMMENT '员工姓名',work_place String COMMENT '工作地点',age UInt8 COMMENT '员工年龄',depart String COMMENT '部门',salary Decimal32(2) COMMENT '工资') ENGINE=StripeLog;
-- 插入数据
INSERT INTO emp_stripelog VALUES (1,'tom','上海',25,'技术部',20000),(2,'jack','上海',26,'人事部',10000);
INSERT INTO emp_stripelog VALUES (3,'bob','北京',33,'财务部',50000),(4,'tony','杭州',28,'销售事部',50000);
-- 查询数据
-- 由于是分两次插入数据,所以查询时会有两个数据块
select * from emp_stripelog;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │
│ 2 │ jack │ 上海 │ 26 │ 人事部 │ 10000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart───┬───salary─┐
│ 3 │ bob │ 北京 │ 33 │ 财务部 │ 50000.00 │
│ 4 │ tony │ 杭州 │ 28 │ 销售事部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴──────────┴──────────┘
进入默认数据存储目录,查看底层数据存储形式
[root@hadoop01 ~]# ll /data/clickhouse/clickhousedata/data/default/emp_stripelog/
total 12
-rw-r----- 1 root root 673 Sep 24 01:07 data.bin
-rw-r----- 1 root root 281 Sep 24 01:07 index.mrk
-rw-r----- 1 root root 69 Sep 24 01:07 sizes.json
可以看出StripeLog表引擎对应的存储结构包括三个文件:
- data.bin:数据文件,所有的列字段使用同一个文件保存,它们的数据都会被写入data.bin。
- index.mrk:数据标记,保存了数据在data.bin文件中的位置信息(
每个插入数据块对应列的offset
),利用数据标记能够使用多个线程,以并行的方式读取data.bin内的压缩数据块,从而提升数据查询的性能。 - sizes.json:元数据文件,记录了data.bin和index.mrk大小的信息
注:
1、
StripeLog
引擎将所有数据都存储在了一个文件中,对于每次的INSERT操作,ClickHouse会将数据块追加到表文件的末尾2、StripeLog引擎同样不支持
ALTER UPDATE
和ALTER DELETE
操作
5.3.5 Log表引擎
Log引擎表适用于临时数据,一次性写入、测试场景。Log引擎结合了TinyLog表引擎和StripeLog表引擎的长处,是Log系列引擎中性能最高的表引擎。
CREATE TABLE emp_log (emp_id UInt16 COMMENT '员工id',name String COMMENT '员工姓名',work_place String COMMENT '工作地点',age UInt8 COMMENT '员工年龄',depart String COMMENT '部门',salary Decimal32(2) COMMENT '工资')ENGINE=Log;
INSERT INTO emp_log VALUES (1,'tom','上海',25,'技术部',20000),(2,'jack','上海',26,'人事部',10000);
INSERT INTO emp_log VALUES (3,'bob','北京',33,'财务部',50000),(4,'tony','杭州',28,'销售事部',50000);
-- 查询数据,
-- 由于是分两次插入数据,所以查询时会有两个数据块
select * from emp_log;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │
│ 2 │ jack │ 上海 │ 26 │ 人事部 │ 10000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart───┬───salary─┐
│ 3 │ bob │ 北京 │ 33 │ 财务部 │ 50000.00 │
│ 4 │ tony │ 杭州 │ 28 │ 销售事部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴──────────┴──────────┘
进入默认数据存储目录,查看底层数据存储形式
[root@hadoop01 ~]# ll /data/clickhouse/clickhousedata/data/default/emp_log/
total 32
-rw-r----- 1 root root 56 Sep 24 01:11 age.bin
-rw-r----- 1 root root 97 Sep 24 01:11 depart.bin
-rw-r----- 1 root root 60 Sep 24 01:11 emp_id.bin
-rw-r----- 1 root root 96 Sep 24 01:11 __marks.mrk
-rw-r----- 1 root root 70 Sep 24 01:11 name.bin
-rw-r----- 1 root root 68 Sep 24 01:11 salary.bin
-rw-r----- 1 root root 215 Sep 24 01:11 sizes.json
-rw-r----- 1 root root 80 Sep 24 01:11 work_place.bin
Log引擎的存储结构包含三部分:
- 列.bin:数据文件,数据文件按列单独存储
- __marks.mrk:数据标记,统一保存了数据在各个.bin文件中的位置信息。利用数据标记能够使用多个线程,以并行的方式读取。.bin内的压缩数据块,从而提升数据查询的性能。
- sizes.json:记录了.bin和__marks.mrk大小的信息
注:
Log表引擎会将每一列都存在一个文件中,对于每一次的INSERT操作,都会对应一个数据块
5.4 MergeTree系列引擎
在所有的表引擎中,最为核心的当属MergeTree系列表引擎,这些表引擎拥有最为强大的性能和最广泛的使用场合。对于非MergeTree系列的其他引擎而言,主要用于特殊用途,场景相对有限。而MergeTree系列表引擎是官方主推的存储引擎,支持几乎所有ClickHouse核心功能。
5.4.1 MergeTree表引擎
MergeTree在写入一批数据时,数据总会以数据片段的形式写入磁盘,且数据片段不可修改。为了避免片段过多,ClickHouse会通过后台线程,定期合并这些数据片段,属于相同分区的数据片段会被合成一个新的片段
。这种数据片段往复合并的特点,也正是合并树名称的由来。
MergeTree作为家族系列最基础的表引擎,主要有以下特点:
- 存储的数据按照主键排序:允许创建稀疏索引,从而加快数据查询速度
- 支持分区,可以通过PARTITION KEY语句指定分区字段。
- 支持数据副本
- 支持数据采样
建表语法
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, ...]
- ENGINE:ENGINE = MergeTree(),MergeTree引擎没有参数
- ORDER BY:排序字段。比如ORDER BY (Col1, Col2),值得注意的是,如果没有指定主键,
默认情况下 sorting key(排序字段)即为主键
。如果不需要排序,则可以使用ORDER BY tuple()语法,这样的话,创建的表也就不包含主键。这种情况下,ClickHouse会按照插入的顺序存储数据。必选。 - PARTITION BY:分区字段,可选。
- PRIMARY KEY:指定主键,如果排序字段与主键不一致,可以单独指定主键字段。否则
默认主键是排序字段
。可选。 - SAMPLE BY:采样字段,如果指定了该字段,那么主键中也必须包含该字段。比如
SAMPLE BY intHash32(UserID) ORDER BY (CounterID, EventDate, intHash32(UserID))
。可选。 - TTL:数据的存活时间。在MergeTree中,可以为某个列字段或整张表设置TTL。当时间到达时,如果是列字段级别的TTL,则会删除这一列的数据;如果是表级别的TTL,则会删除整张表的数据。可选。
- SETTINGS:额外的参数配置。可选。
建表示例
CREATE TABLE emp_mergetree (emp_id UInt16 COMMENT '员工id',name String COMMENT '员工姓名',work_place String COMMENT '工作地点',age UInt8 COMMENT '员工年龄',depart String COMMENT '部门',salary Decimal32(2) COMMENT '工资')ENGINE=MergeTree() ORDER BY emp_id PARTITION BY work_place;
-- 插入数据
INSERT INTO emp_mergetree VALUES (1,'tom','上海',25,'技术部',20000),(2,'jack','上海',26,'人事部',10000);
INSERT INTO emp_mergetree VALUES (3,'bob','北京',33,'财务部',50000),(4,'tony','杭州',28,'销售事部',50000);
-- 查询数据
-- 按work_place进行分区
select * from emp_mergetree;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │
│ 2 │ jack │ 上海 │ 26 │ 人事部 │ 10000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart───┬───salary─┐
│ 4 │ tony │ 杭州 │ 28 │ 销售事部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴──────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 3 │ bob │ 北京 │ 33 │ 财务部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
查看一下数据存储格式,可以看出,存在三个分区文件夹,每一个分区文件夹内存储了对应分区的数据。
[root@hadoop01 ~]# ll /data/clickhouse/clickhousedata/data/default/emp_mergetree/
total 16
drwxr-x--- 2 root root 4096 Sep 24 01:17 1c89a3ba9fe5fd53379716a776c5ac34_3_3_0
drwxr-x--- 2 root root 4096 Sep 24 01:17 40d45822dbd7fa81583d715338929da9_1_1_0
drwxr-x--- 2 root root 4096 Sep 24 01:17 a6155dcc1997eda1a348cd98b17a93e9_2_2_0
drwxr-x--- 2 root root 6 Sep 24 01:15 detached
-rw-r----- 1 root root 1 Sep 24 01:15 format_version.txt
进入一个分区目录查看
-
checksums.txt:校验文件,使用二进制格式存储。它保存了余下各类文件(primary. idx、count.txt等)的size大小及size的哈希值,用于快速校验文件的完整性和正确性。
-
columns.txt:列信息文件,使用明文格式存储。用于保存此数据分区下的列字段信息,例如
[root@hadoop01 ~]# cat /data/clickhouse/clickhousedata/data/default/emp_mergetree/1c89a3ba9fe5fd53379716a776c5ac34_3_3_0/columns.txt columns format version: 1 6 columns: `emp_id` UInt16 `name` String `work_place` String `age` UInt8 `depart` String `salary` Decimal(9, 2)
-
count.txt:计数文件,使用明文格式存储。用于记录当前数据分区目录下数据的总行数
-
primary.idx:一级索引文件,使用二进制格式存储。用于存放稀疏索引,一张MergeTree表只能声明一次一级索引,即通过ORDER BY或者PRIMARY KEY指定字段。借助稀疏索引,在数据查询的时能够排除主键条件范围之外的数据文件,从而有效减少数据扫描范围,加速查询速度。
-
列.bin:数据文件,使用压缩格式存储,默认为LZ4压缩格式,用于存储某一列的数据。由于MergeTree采用列式存储,所以每一个列字段都拥有独立的
.bin
数据文件,并以列字段名称命名。 -
列.mrk2:列字段标记文件,使用二进制格式存储。标记文件中保存了
.bin
文件中数据的偏移量信息 -
partition.dat与minmax_[Column].idx:如果指定了分区键,则会额外生成partition.dat与minmax索引文件,它们均使用二进制格式存储。
**partition.dat**用于保存当前分区下分区表达式最终生成的值
,即分区字段值;而minmax索引用于记录当前分区下分区字段对应原始数据的最小和最大值。比如当使用EventTime字段对应的原始数据为2020-09-17、2020-09-30,分区表达式为PARTITION BY toYYYYMM(EventTime),即按月分区。partition.dat中保存的值将会是2019-09,而minmax索引中保存的值将会是2020-09-17 2020-09-30。
注意点
多次插入数据,会生成多个分区文件
-- 新插入两条数据
INSERT INTO emp_mergetree VALUES (5,'robin','北京',35,'财务部',50000),(6,'lilei','北京',38,'销售事部',50000);
-- 查询结果
select * from emp_mergetree;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 3 │ bob │ 北京 │ 33 │ 财务部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name──┬─work_place─┬─age─┬─depart───┬───salary─┐
│ 5 │ robin │ 北京 │ 35 │ 财务部 │ 50000.00 │
│ 6 │ lilei │ 北京 │ 38 │ 销售事部 │ 50000.00 │
└────────┴───────┴────────────┴─────┴──────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart───┬───salary─┐
│ 4 │ tony │ 杭州 │ 28 │ 销售事部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴──────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │
│ 2 │ jack │ 上海 │ 26 │ 人事部 │ 10000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
可以看出,新插入的数据新生成了一个数据块,并没有与原来的分区数据在一起,我们可以执行optimize命令,执行合并操作
-- 执行合并操作
hadoop01 :) OPTIMIZE TABLE emp_mergetree PARTITION '北京';
-- 再次执行查询
select * from emp_mergetree;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │
│ 2 │ jack │ 上海 │ 26 │ 人事部 │ 10000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name──┬─work_place─┬─age─┬─depart───┬───salary─┐
│ 3 │ bob │ 北京 │ 33 │ 财务部 │ 50000.00 │
│ 5 │ robin │ 北京 │ 35 │ 财务部 │ 50000.00 │
│ 6 │ lilei │ 北京 │ 38 │ 销售事部 │ 50000.00 │
└────────┴───────┴────────────┴─────┴──────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart───┬───salary─┐
│ 4 │ tony │ 杭州 │ 28 │ 销售事部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴──────────┴──────────┘
执行上面的合并操作之后,会新生成一个该分区的文件夹,原理的分区文件夹不变。
- 在MergeTree中
主键并不用于去重,而是用于索引
,加快查询速度
-- 插入一条相同主键的数据
INSERT INTO emp_mergetree VALUES (1,'sam','杭州',35,'财务部',50000);
-- 会发现该条数据可以插入,由此可知,并不会对主键进行去重
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │
│ 2 │ jack │ 上海 │ 26 │ 人事部 │ 10000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name──┬─work_place─┬─age─┬─depart───┬───salary─┐
│ 3 │ bob │ 北京 │ 33 │ 财务部 │ 50000.00 │
│ 5 │ robin │ 北京 │ 35 │ 财务部 │ 50000.00 │
│ 6 │ lilei │ 北京 │ 38 │ 销售事部 │ 50000.00 │
└────────┴───────┴────────────┴─────┴──────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart───┬───salary─┐
│ 4 │ tony │ 杭州 │ 28 │ 销售事部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴──────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ sam │ 杭州 │ 35 │ 财务部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
5.4.2 ReplacingMergeTree表引擎
上文提到MergeTree表引擎无法对相同主键(排序键
)的数据进行去重,ClickHouse提供了ReplacingMergeTree引擎,可以针对相同主键的数据进行去重,它能够在合并分区时删除重复的数据。
值得注意的是,ReplacingMergeTree只是在一定程度上解决了数据重复问题,但是并不能完全保障数据不重复。
建表语法
CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
...
) ENGINE = ReplacingMergeTree([ver])
[PARTITION BY expr]
[ORDER BY expr]
[PRIMARY KEY expr]
[SAMPLE BY expr]
[SETTINGS name=value, ...]
- [ver]:可选参数列的版本,可以是UInt、Date或者DateTime类型的字段作为版本号。该参数决定了数据去重的方式。
当没有指定[ver]参数时,保留最新的数据;如果指定了具体的值,保留最大的版本数据
。
建表示例
CREATE TABLE emp_replacingmergetree (emp_id UInt16 COMMENT '员工id',name String COMMENT '员工姓名',work_place String COMMENT '工作地点',age UInt8 COMMENT '员工年龄',depart String COMMENT '部门',salary Decimal32(2) COMMENT '工资')ENGINE=ReplacingMergeTree() ORDER BY emp_id PRIMARY KEY emp_id PARTITION BY work_place;
-- 插入数据
INSERT INTO emp_replacingmergetree VALUES (1,'tom','上海',25,'技术部',20000),(2,'jack','上海',26,'人事部',10000);
INSERT INTO emp_replacingmergetree VALUES (3,'bob','北京',33,'财务部',50000),(4,'tony','杭州',28,'销售事部',50000);
select * from emp_replacingmergetree;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │
│ 2 │ jack │ 上海 │ 26 │ 人事部 │ 10000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart───┬───salary─┐
│ 4 │ tony │ 杭州 │ 28 │ 销售事部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴──────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 3 │ bob │ 北京 │ 33 │ 财务部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
注意点
当我们再次向该表插入具有相同主键的数据时,观察查询数据的变化
INSERT INTO emp_replacingmergetree VALUES (1,'tom','上海',25,'技术部',50000);
-- 查询数据,由于没有进行合并,所以存在主键重复的数据
select * from emp_replacingmergetree;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart───┬───salary─┐
│ 4 │ tony │ 杭州 │ 28 │ 销售事部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴──────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 3 │ bob │ 北京 │ 33 │ 财务部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │
│ 2 │ jack │ 上海 │ 26 │ 人事部 │ 10000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
-- 执行合并操作
optimize table emp_replacingmergetree final;
-- 再次查询,相同主键的数据,保留最近插入的数据,旧的数据被清除
select * from emp_replacingmergetree;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 50000.00 │
│ 2 │ jack │ 上海 │ 26 │ 人事部 │ 10000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart───┬───salary─┐
│ 4 │ tony │ 杭州 │ 28 │ 销售事部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴──────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 3 │ bob │ 北京 │ 33 │ 财务部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
从上面的示例中可以看出,ReplacingMergeTree是支持对数据去重的,那么是根据什么进行去重呢?答案是:**ReplacingMergeTree在去除重复数据时,是以ORDERBY排序键为基准的,而不是PRIMARY KEY**
。我们在看一个示例:
CREATE TABLE emp_replacingmergetree1 (emp_id UInt16 COMMENT '员工id',name String COMMENT '员工姓名',work_place String COMMENT '工作地点',age UInt8 COMMENT '员工年龄',depart String COMMENT '部门',salary Decimal32(2) COMMENT '工资')ENGINE=ReplacingMergeTree() ORDER BY (emp_id,name) PRIMARY KEY emp_id PARTITION BY work_place;
ORDER BY (emp_id,name) -- 注意排序key是两个字段
PRIMARY KEY emp_id -- 主键是一个字段
-- 插入数据
INSERT INTO emp_replacingmergetree1 VALUES (1,'tom','上海',25,'技术部',20000),(2,'jack','上海',26,'人事部',10000);
INSERT INTO emp_replacingmergetree1 VALUES (3,'bob','北京',33,'财务部',50000),(4,'tony','杭州',28,'销售事部',50000);
再次向该表中插入相同emp_id和name的数据,并执行合并操作,再观察数据
-- 插入数据
INSERT INTO emp_replacingmergetree1 VALUES (1,'tom','上海',25,'技术部',50000),(1,'sam','上海',25,'技术部',20000);
-- 执行合并操作
optimize table emp_replacingmergetree1 final;
-- 再次查询,可见相同的emp_id和name数据被去重,而相同的主键emp_id不会去重
-- ReplacingMergeTree在去除重复数据时,是以ORDER BY排序键为基准的,而不是PRIMARY KEY
select * from emp_replacingmergetree1;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ sam │ 上海 │ 25 │ 技术部 │ 20000.00 │
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 50000.00 │
│ 2 │ jack │ 上海 │ 26 │ 人事部 │ 10000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 3 │ bob │ 北京 │ 33 │ 财务部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart───┬───salary─┐
│ 4 │ tony │ 杭州 │ 28 │ 销售事部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴──────────┴──────────┘
至此,我们知道了ReplacingMergeTree是支持去重的,并且是按照ORDERBY排序键为基准进行去重的。细心的你会发现,上面的重复数据是在一个分区内的,那么如果重复的数据``不在一个分区内,会发生什么现象呢
?我们再次向上面的emp_replacingmergetree1表插入不同分区的重复数据
-- 插入数据
INSERT INTO emp_replacingmergetree1 VALUES (1,'tom','北京',26,'技术部',10000);
-- 执行合并操作
optimize table emp_replacingmergetree1 final;
-- 再次查询
-- 发现 1 │ tom │ 北京 │ 26 │ 技术部 │ 10000.00
-- 与 1 │ tom │ 上海 │ 25 │ 技术部 │ 50000.00
-- 数据重复,因为这两行数据不在同一个分区内
-- 这是因为ReplacingMergeTree是以分区为单位删除重复数据的。
-- 只有在相同的数据分区内重复的数据才可以被删除,而不同数据分区之间的重复数据依然不能被剔除
select * from emp_replacingmergetree1;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart───┬───salary─┐
│ 4 │ tony │ 杭州 │ 28 │ 销售事部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴──────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ sam │ 上海 │ 25 │ 技术部 │ 20000.00 │
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 50000.00 │
│ 2 │ jack │ 上海 │ 26 │ 人事部 │ 10000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 北京 │ 26 │ 技术部 │ 10000.00 │
│ 3 │ bob │ 北京 │ 33 │ 财务部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
总结
- 如何判断数据重复
ReplacingMergeTree在去除重复数据时,是以ORDERBY排序键为基准的,而不是PRIMARY KEY
。
- 何时删除重复数据
在执行分区合并时,会触发删除重复数据。optimize的合并操作是在后台执行的,无法预测具体执行时间点,除非是手动执行。
- 不同分区的重复数据不会被去重
ReplacingMergeTree是以分区为单位删除重复数据的。只有在相同的数据分区内重复的数据才可以被删除
,而不同数据分区之间的重复数据依然不能被剔除。
- 数据去重的策略是什么
如果没有设置[ver]版本号(ver为数字,日期类型字段),则保留同一组重复数据中的最新插入的数据;如果设置了[ver]版本号,则保留同一组重复数据中ver字段取值最大的那一行。
- optimize命令使用
一般在数据量比较大的情况,尽量不要使用该命令。因为在海量数据场景下,执行optimize要消耗大量时间
5.4.3 SummingMergeTree表引擎
按排序键汇总指定列。该引擎继承了MergeTree引擎,当合并 SummingMergeTree
表的数据片段时,ClickHouse 会把所有具有相同排序键的行合并为一行
,该行包含了被合并的行中具有数值数据类型的列的汇总值,即如果存在重复的数据,会对对这些重复的数据进行合并成一条数据,类似于group by的效果。
推荐将该引擎和
MergeTree 一起使用
。例如,将完整的数据存储在 MergeTree
表中,并且使用 SummingMergeTree
来存储聚合数据。这种方法可以避免因为使用不正确的主键组合方式而丢失数据。
如果用户只需要查询数据的汇总结果,不关心明细数据,并且数据的汇总条件是预先明确的,即GROUP BY的分组字段是确定的,可以使用该表引擎。
建表语法
CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
...
) ENGINE = SummingMergeTree([columns]) -- 指定合并汇总字段
[PARTITION BY expr]
[ORDER BY expr]
[SAMPLE BY expr]
[SETTINGS name=value, ...]
建表示例
CREATE TABLE emp_summingmergetree (emp_id UInt16 COMMENT '员工id',name String COMMENT '员工姓名',work_place String COMMENT '工作地点',age UInt8 COMMENT '员工年龄',depart String COMMENT '部门',salary Decimal32(2) COMMENT '工资')
ENGINE=SummingMergeTree(salary)
ORDER BY (emp_id,name)
PRIMARY KEY emp_id
PARTITION BY work_place;
ORDER BY (emp_id,name) -- 注意排序key是两个字段
PRIMARY KEY emp_id -- 主键是一个字段
-- 插入数据
INSERT INTO emp_summingmergetree VALUES (1,'tom','上海',25,'技术部',20000),(2,'jack','上海',26,'人事部',10000);
INSERT INTO emp_summingmergetree VALUES (3,'bob','北京',33,'财务部',50000),(4,'tony','杭州',28,'销售事部',50000);
当我们再次插入具有相同emp_id,name的数据时,观察结果
INSERT INTO emp_summingmergetree VALUES (1,'tom','上海',25,'信息部',10000),(1,'tom','北京',26,'人事部',10000);
select * from emp_summingmergetree;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 上海 │ 25 │ 信息部 │ 10000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart───┬───salary─┐
│ 4 │ tony │ 杭州 │ 28 │ 销售事部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴──────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 3 │ bob │ 北京 │ 33 │ 财务部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │
│ 2 │ jack │ 上海 │ 26 │ 人事部 │ 10000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 北京 │ 26 │ 人事部 │ 10000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
-- 执行合并操作
optimize table emp_summingmergetree final;
select * from emp_summingmergetree;
-- 再次查询,新插入的数据 1 │ tom │ 上海 │ 25 │ 信息部 │ 10000.00
-- 原来的数据 : 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00
-- 这两行数据合并成: 1 │ tom │ 上海 │ 25 │ 技术部 │ 30000.00
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 30000.00 │
│ 2 │ jack │ 上海 │ 26 │ 人事部 │ 10000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart───┬───salary─┐
│ 4 │ tony │ 杭州 │ 28 │ 销售事部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴──────────┴──────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 北京 │ 26 │ 人事部 │ 10000.00 │
│ 3 │ bob │ 北京 │ 33 │ 财务部 │ 50000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
注意点
要保证**PRIMARY KEY expr**指定的主键是**ORDER BY expr** 指定字段的前缀
,比如
-- 允许
ORDER BY (A,B,C)
PRIMARY KEY A
-- 会报错
-- DB::Exception: Primary key must be a prefix of the sorting key
ORDER BY (A,B,C)
PRIMARY KEY B
这种强制约束保障了即便在两者定义不同的情况下,主键仍然是排序键的前缀,不会出现索引与数据顺序混乱
的问题。
总结
- SummingMergeTree是根据什么对两条数据进行合并的
用ORBER BY排序键作为聚合数据的条件Key
。即如果排序key是相同的,则会合并成一条数据,并对指定的合并字段进行聚合。
- 仅对分区内的相同排序key的数据行进行合并
以数据分区为单位来聚合数据。当分区合并时,同一数据分区内聚合Key相同的数据会被合并汇总,而不同分区之间的数据则不会被汇总。
- 如果没有指定聚合字段,会怎么聚合
如果没有指定聚合字段ENGINE=SummingMergeTree(salary),则会按照非主键的数值类型字段进行聚合
- 对于非汇总字段的数据,该保留哪一条
如果两行数据除了排序字段相同,其他的非聚合字段不相同,那么在聚合发生时,会保留最初的那条数据,新插入的数据对应的那个字段值会被舍弃
。下面新插入的“信息部
”被舍弃
-- 新插入的数据: 1 │ tom │ 上海 │ 25 │ 信息部 │ 10000.00
-- 最初的数据 : 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00
-- 聚合合并的结果: 1 │ tom │ 上海 │ 25 │ 技术部 │ 30000.00
5.4.4 AggregatingMergetree表引擎
按排序键自定义聚合。
该表引擎继承自MergeTree,可以使用 AggregatingMergeTree
表来做增量数据统计聚合。如果要按一组规则来合并减少行数,则使用 AggregatingMergeTree
是合适的。AggregatingMergeTree是通过预先定义的聚合函数计算数据并通过二进制的格式存入表
内。
与SummingMergeTree的区别在于:SummingMergeTree对非主键列进行sum聚合,而AggregatingMergeTree则可以指定各种聚合函数
。
建表语法
CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
...
) ENGINE = AggregatingMergeTree()
[PARTITION BY expr]
[ORDER BY expr]
[SAMPLE BY expr]
[SETTINGS name=value, ...]
建表示例
CREATE TABLE emp_aggregatingmergeTree (emp_id UInt16 COMMENT '员工id',name String COMMENT '员工姓名',work_place String COMMENT '工作地点',age UInt8 COMMENT '员工年龄',depart String COMMENT '部门',salary AggregateFunction(sum,Decimal32(2)) COMMENT '工资')ENGINE=AggregatingMergeTree() ORDER BY (emp_id,name) PRIMARY KEY emp_id PARTITION BY work_place;
ORDER BY (emp_id,name) -- 注意排序key是两个字段
PRIMARY KEY emp_id -- 主键是一个字段
对于AggregateFunction类型
的列字段,在进行数据的写入和查询时与其他的表引擎有很大区别,在写入数据时,需要调用-State函数(聚合函数组合器https://clickhouse.tech/docs/zh/sql-reference/aggregate-functions/combinators/#aggregate_functions_combinators-merge);而在查询数据时,则需要调用相应的-Merge函数。对于上面的建表语句而言,需要使用**sumState**
函数进行数据插入
,使用sumMerge
函数查询
。
salary AggregateFunction(sum,Decimal32(2)) COMMENT '工资'
AggregateFunction(...)
类型可用于进一步处理或存储在表中,以完成稍后的聚合。
sum:汇总聚合函数,sumState(toDecimal32(10000,2))转为小数并存储为二进制的AggregateFunction类型,sumMerge(salary)将中间聚合状态作为参数,组合状态以完成聚合,并返回结果值。
Decimal32(2):表示该列存储数值类型精度为2
-- 插入数据,
-- 注意:需要使用INSERT…SELECT语句进行数据插入
INSERT INTO TABLE emp_aggregatingmergeTree SELECT 1,'tom','上海',25,'信息部',sumState(toDecimal32(10000,2));
INSERT INTO TABLE emp_aggregatingmergeTree SELECT 1,'tom','上海',25,'信息部',sumState(toDecimal32(20000,2));
-- 查询数据
SELECT emp_id,name,sumMerge(salary) FROM emp_aggregatingmergeTree GROUP BY emp_id,name;
-- 结果输出
┌─emp_id─┬─name─┬─sumMerge(salary)─┐
│ 1 │ tom │ 30000.00 │
└────────┴──────┴──────────────────┘
上面演示的用法非常的麻烦,其实更多的情况下,我们可以结合物化视图一起使用,将它作为物化视图(存储数据)的表引擎。而这里的物化视图是作为其他数据表上层的一种查询视图。
AggregatingMergeTree通常作为物化视图的表引擎,与普通MergeTree搭配使用。
-- 创建一个MereTree引擎的明细表
-- 用于存储全量的明细数据
-- 对外提供实时查询
CREATE TABLE emp_mergetree_base (emp_id UInt16 COMMENT '员工id',name String COMMENT '员工姓名',work_place String COMMENT '工作地点',age UInt8 COMMENT '员工年龄',depart String COMMENT '部门',salary Decimal32(2) COMMENT '工资')ENGINE=MergeTree() ORDER BY (emp_id,name) PARTITION BY work_place;
-- 创建一张物化视图
-- 使用AggregatingMergeTree表引擎
CREATE MATERIALIZED VIEW view_emp_agg
ENGINE = AggregatingMergeTree()
PARTITION BY emp_id
ORDER BY (emp_id,name)
AS
SELECT emp_id,name,sumState(salary) AS salary FROM emp_mergetree_base GROUP BY emp_id,name;
-- 向基础明细表emp_mergetree_base插入数据
INSERT INTO emp_mergetree_base VALUES (1,'tom','上海',25,'技术部',20000),(1,'tom','上海',26,'人事部',10000);
-- 查询物化视图
SELECT emp_id,name,sumMerge(salary) FROM view_emp_agg GROUP BY emp_id,name;
-- 结果
┌─emp_id─┬─name─┬─sumMerge(salary)─┐
│ 1 │ tom │ 50000.00 │
└────────┴──────┴──────────────────┘
SimpleAggregateFunction
SimpleAggregateFunction(name, types_of_arguments…)
数据类型存储聚合函数的当前值,而不将其完整状态存储为 AggregateFunction
有 此优化可应用于具有以下属性的函数:应用函数的结果 f
到行集 S1 UNION ALL S2
可以通过应用来获得 f
行的部分单独设置,然后再次应用 f
到结果: f(S1 UNION ALL S2) = f(f(S1) UNION ALL f(S2))
. 此属性保证部分聚合结果足以计算组合结果,因此我们不必存储和处理任何额外的数据。
支持以下聚合函数:
的值 SimpleAggregateFunction(func, Type)
看起来和存储方式相同 Type
,所以你不需要应用函数 -Merge
/-State
后缀。 SimpleAggregateFunction
具有比更好的性能 AggregateFunction
具有相同的聚合功能。
参数
- 聚合函数的名称。
- 聚合函数参数的类型。
5.5.5 CollapsingMergeTree表引擎
折叠合并树,CollapsingMergeTree就是一种通过以增代删
的思路,支持行级数据修改和删除的表引擎。它通过定义一个sign标记位字段
,记录数据行的状态。
如果sign标记为1,则表示这是一行有效的数据;
如果sign标记为-1,则表示这行数据需要被删除。
当CollapsingMergeTree分区合并时,同一数据分区内,sign标记为1和-1的一组数据会被抵消删除。
每次需要新增数据时,写入一行sign标记为1的数据;需要删除数据时,则写入一行sign标记为-1的数据。
建表语法
CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
...
) ENGINE = CollapsingMergeTree(sign)
[PARTITION BY expr]
[ORDER BY expr]
[SAMPLE BY expr]
[SETTINGS name=value, ...]
建表示例
上面的建表语句使用CollapsingMergeTree(sign),其中字段sign是一个Int8类型的字段
CREATE TABLE emp_collapsingmergetree (emp_id UInt16 COMMENT '员工id',name String COMMENT '员工姓名',work_place String COMMENT '工作地点',age UInt8 COMMENT '员工年龄',depart String COMMENT '部门',salary Decimal32(2) COMMENT '工资',sign Int8)ENGINE=CollapsingMergeTree(sign) ORDER BY (emp_id,name) PARTITION BY work_place;
使用方式
CollapsingMergeTree同样是以ORDER BY排序键作为判断数据唯一性的依据。
-- 插入新增数据,sign=1表示正常数据
INSERT INTO emp_collapsingmergetree VALUES (1,'tom','上海',25,'技术部',20000,1);
-- 更新上述的数据
-- 首先插入一条与原来相同的数据(ORDER BY字段一致),并将sign置为-1
INSERT INTO emp_collapsingmergetree VALUES (1,'tom','上海',25,'技术部',20000,-1);
-- 再插入更新之后的数据
INSERT INTO emp_collapsingmergetree VALUES (1,'tom','上海',25,'技术部',30000,1);
-- 查看一下结果
select * from emp_collapsingmergetree ;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┬─sign─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │ -1 │
└────────┴──────┴────────────┴─────┴────────┴──────────┴──────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┬─sign─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │ 1 │
└────────┴──────┴────────────┴─────┴────────┴──────────┴──────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┬─sign─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 30000.00 │ 1 │
└────────┴──────┴────────────┴─────┴────────┴──────────┴──────┘
-- 执行分区合并操作
optimize table emp_collapsingmergetree;
-- 再次查询,sign=1与sign=-1的数据相互抵消了,即被删除
select * from emp_collapsingmergetree ;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┬─sign─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 30000.00 │ 1 │
└────────┴──────┴────────────┴─────┴────────┴──────────┴──────┘
注意点
- 分区合并
分区数据折叠不是实时的,需要后台进行Compaction操作,用户也可以使用手动合并命令,但是效率会很低,一般不推荐在生产环境中使用。
当进行汇总数据操作时,可以通过改变查询方式,来过滤掉被删除的数据
SELECT emp_id,name,sum(salary * sign) FROM emp_collapsingmergetree GROUP BY emp_id, name HAVING sum(sign) > 0;
┌─emp_id─┬─name─┬─sum(multiply(salary, sign))─┐
│ 1 │ tom │ 30000.00 │
└────────┴──────┴─────────────────────────────┘
-- sum(salary * sign) 删除数据是1和-1,这样的数据sum(salary*sign)刚好抵消。
-- HAVING sum(sign) > 0 确保过滤掉sign仅有1和-1的数据
只有相同分区内的数据才有可能被折叠
。其实,当我们修改或删除数据时,这些被修改的数据通常是在一个分区内的,所以不会产生影响。
- 数据写入顺序
值得注意的是:CollapsingMergeTree对于写入数据的顺序有着严格要求:先加后删
(必须先sing=1,再sing=-1的顺序才能删除;反之则不能删除),否则导致无法正常折叠。
-- 建表
CREATE TABLE emp_collapsingmergetree_order (emp_id UInt16 COMMENT '员工id',name String COMMENT '员工姓名',work_place String COMMENT '工作地点',age UInt8 COMMENT '员工年龄',depart String COMMENT '部门',salary Decimal32(2) COMMENT '工资',sign Int8)ENGINE=CollapsingMergeTree(sign) ORDER BY (emp_id,name) PARTITION BY work_place;
-- 先插入需要被删除的数据,即sign=-1的数据
INSERT INTO emp_collapsingmergetree_order VALUES (1,'tom','上海',25,'技术部',20000,-1);
-- 再插入sign=1的数据
INSERT INTO emp_collapsingmergetree_order VALUES (1,'tom','上海',25,'技术部',20000,1);
-- 查询表
SELECT * FROM emp_collapsingmergetree_order;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┬─sign─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │ -1 │
└────────┴──────┴────────────┴─────┴────────┴──────────┴──────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┬─sign─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │ 1 │
└────────┴──────┴────────────┴─────┴────────┴──────────┴──────┘
-- 执行合并操作
optimize table emp_collapsingmergetree_order;
-- 再次查询表
-- 旧数据依然存在
SELECT * FROM emp_collapsingmergetree_order;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┬─sign─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │ -1 │
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │ 1 │
└────────┴──────┴────────────┴─────┴────────┴──────────┴──────┘
如果数据的写入程序是单线程执行的,则能够较好地控制写入顺序;
如果需要处理的数据量很大,数据的写入程序通常是多线程执行的,那么此时就不能保障数据的写入顺序了。在这种情况下,CollapsingMergeTree的工作机制就会出现问题。但是可以通过VersionedCollapsingMergeTree的表引擎得到解决。
5.5.6 VersionedCollapsingMergeTree表引擎
版本折叠合并树。上面提到CollapsingMergeTree表引擎对于数据写入乱序的情况下,不能够实现数据折叠的效果。VersionedCollapsingMergeTree表引擎的作用与CollapsingMergeTree完全相同,它们的不同之处在于,VersionedCollapsingMergeTree对数据的写入顺序没有要求
,在同一个分区内,任意顺序的数据都能够完成折叠操作。
VersionedCollapsingMergeTree使用version列来实现乱序情况下的数据折叠。
建表语法
CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
...
) ENGINE = VersionedCollapsingMergeTree(sign, version)
[PARTITION BY expr]
[ORDER BY expr]
[SAMPLE BY expr]
[SETTINGS name=value, ...]
可以看出:该引擎除了需要指定一个sign标识之外,还需要指定一个UInt8类型的version版本号
。
建表示例
CREATE TABLE emp_versioned (emp_id UInt16 COMMENT '员工id',name String COMMENT '员工姓名',work_place String COMMENT '工作地点',age UInt8 COMMENT '员工年龄',depart String COMMENT '部门',salary Decimal32(2) COMMENT '工资',sign Int8,version Int8)ENGINE=VersionedCollapsingMergeTree(sign, version) ORDER BY (emp_id,name) PARTITION BY work_place;
-- 先插入需要被删除的数据,即sign=-1的数据
INSERT INTO emp_versioned VALUES (1,'tom','上海',25,'技术部',20000,-1,1);
-- 再插入sign=1的数据
INSERT INTO emp_versioned VALUES (1,'tom','上海',25,'技术部',20000,1,1);
-- 在插入一个新版本数据
INSERT INTO emp_versioned VALUES (1,'tom','上海',25,'技术部',30000,1,2);
-- 先不执行合并,查看表数据
select * from emp_versioned;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┬─sign─┬─version─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │ 1 │ 1 │
└────────┴──────┴────────────┴─────┴────────┴──────────┴──────┴─────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┬─sign─┬─version─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │ -1 │ 1 │
└────────┴──────┴────────────┴─────┴────────┴──────────┴──────┴─────────┘
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┬─sign─┬─version─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 30000.00 │ 1 │ 2 │
└────────┴──────┴────────────┴─────┴────────┴──────────┴──────┴─────────┘
-- 获取正确查询结果
SELECT emp_id,name,sum(salary * sign) FROM emp_versioned GROUP BY emp_id,name HAVING sum(sign) > 0;
┌─emp_id─┬─name─┬─sum(multiply(salary, sign))─┐
│ 1 │ tom │ 30000.00 │
└────────┴──────┴─────────────────────────────┘
-- 手动合并
optimize table emp_versioned;
-- 再次查询
select * from emp_versioned;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┬─sign─┬─version─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 30000.00 │ 1 │ 2 │
└────────┴──────┴────────────┴─────┴────────┴──────────┴──────┴─────────┘
可见上面虽然在插入数据乱序的情况下,依然能够实现折叠的效果。之所以能够达到这种效果,是因为在定义version字段之后,VersionedCollapsingMergeTree会自动将version作为排序条件并增加到ORDER BY的末端,就上述的例子而言,最终的排序字段为ORDER BY emp_id,name,version desc
。
5.5 外部集成表引擎
ClickHouse提供了许多与外部系统集成的方法,包括一些表引擎。这些表引擎与其他类型的表引擎类似,可以用于将外部数据导入到ClickHouse中,或者在ClickHouse中直接操作外部数据源。
例如直接读取HDFS的文件或者MySQL数据库的表。这些表引擎只负责元数据管理和数据查询,而它们自身通常并不负责数据的写入,数据文件直接由外部系统提供。目前ClickHouse提供了下面的外部集成表引擎:
- ODBC:通过指定odbc连接读取数据源
- JDBC:通过指定jdbc连接读取数据源;
- MySQL:将MySQL作为数据存储,直接查询其数据
- HDFS:直接读取HDFS上的特定格式的数据文件;
- Kafka:将Kafka数据导入ClickHouse
- RabbitMQ:与Kafka类似
5.5.1 HDFS(hadoop 2.6兼容性差)
使用方式
ENGINE = HDFS(URI, format)
- URI:HDFS文件路径
- format:文件格式,比如CSV、JSON、TSV等
使用示例
-- 建表
CREATE TABLE hdfs_engine_table(emp_id UInt16 COMMENT '员工id',name String COMMENT '员工姓名',work_place String COMMENT '工作地点',age UInt8 COMMENT '员工年龄',depart String COMMENT '部门',salary Decimal32(2) COMMENT '工资') ENGINE=HDFS('hdfs://hadoop01/user/hive/hdfs_engine_table', 'CSV');
-- 写入数据
INSERT INTO hdfs_engine_table VALUES (1,'tom','上海',25,'技术部',20000),(2,'jack','上海',26,'人事部',10000);
-- 查询数据
select * from hdfs_engine_table;
┌─emp_id─┬─name─┬─work_place─┬─age─┬─depart─┬───salary─┐
│ 1 │ tom │ 上海 │ 25 │ 技术部 │ 20000.00 │
│ 2 │ jack │ 上海 │ 26 │ 人事部 │ 10000.00 │
└────────┴──────┴────────────┴─────┴────────┴──────────┘
--再在HDFS上其对应的文件,添加几条数据,再次查看
select * from hdfs_engine_table;
可以看出,这种方式与使用Hive类似,我们直接可以将HDFS对应的文件映射成ClickHouse中的一张表,这样就可以使用SQL操作HDFS上的文件了。
值得注意的是:ClickHouse并不能够删除HDFS上的数据,当我们在ClickHouse客户端中删除了对应的表,只是删除了表结构,HDFS上的文件并没有被删除,这一点跟Hive的外部表十分相似。
5.5.2 MySQL
在第四章中介绍了MySQL数据库引擎,即ClickHouse可以创建一个MySQL数据引擎,这样就可以在ClickHouse中操作其对应的数据库中的数据。其实,ClickHouse同样支持MySQL表引擎,即映射一张MySQL中的表到ClickHouse中。
注意:对于MySQL表引擎,不支持UPDATE和DELETE操作,比如执行下面命令时,会报错:
-- 执行更新
ALTER TABLE mysql_engine_table UPDATE name = 'hanmeimei' WHERE id = 1;
-- 执行删除
ALTER TABLE mysql_engine_table DELETE WHERE id = 1;
-- 报错
DB::Exception: Mutations are not supported by storage MySQL.
5.5.3 JDBC
使用方式
JDBC表引擎不仅可以对接MySQL数据库,还能够与PostgreSQL等数据库。为了实现JDBC连接,ClickHouse使用了clickhouse-jdbc-bridge的查询代理服务。
1、首先我们需要下载clickhouse-jdbc-bridge(github地址:https://github.com/ClickHouse/clickhouse-jdbc-bridge),然后按照ClickHouse的github中的步骤进行编译,编译完成之后会有一个clickhouse-jdbc-bridge-1.0.jar的jar文件,除了需要该文件之外,还需要JDBC的驱动文件,本文使用的是MySQL,所以还需要下载MySQL驱动包。将MySQL的驱动包和clickhouse-jdbc-bridge-1.0.jar文件放在了/usr/local/clickhouse/softwares路径下,执行如下命令:
[root@hadoop01 home]# java -jar clickhouse-jdbc-bridge-1.0.jar --driver-path /usr/local/clickhouse/softwares --listen-host hadoop01
其中--driver-path
是MySQL驱动的jar所在的路径,listen-host
是代理服务绑定的主机。默认情况下,绑定的端口是:9019。
2、然后我们再配置/etc/clickhouse-server/config.xml
,在文件中添加如下配置,然后重启服务。
<jdbc_bridge>
<host>hadoop01</host>
<port>9019</port>
</jdbc_bridge>
使用示例
- 直接查询MySQL中对应的表
SELECT * FROMjdbc('jdbc:mysql://192.168.216.111:3306/?user=root&password=root', 'sales_source','product');
- 创建一张映射表
-- 语法
CREATE TABLE [IF NOT EXISTS] [db.]table_name
(
columns list...
)
ENGINE = JDBC(dbms_uri, external_database, external_table)
-- MySQL建表
CREATE TABLE jdbc_table_mysql (order_id INT NOT NULL AUTO_INCREMENT,amount FLOAT NOT NULL,PRIMARY KEY (order_id));
-- MySQL插入数据
INSERT INTO jdbc_table_mysql VALUES (1,200);
-- 在ClickHouse中建表
CREATE TABLE jdbc_table(order_id Int32,amount Float32) ENGINE JDBC('jdbc:mysql://192.168.216.111:3306/?user=root&password=root','sales_source','jdbc_table_mysql');
-- clickhouse查询数据
select * from jdbc_table;
┌─order_id─┬─amount─┐
│ 1 │ 200 │
└──────────┴────────┘
5.5.4 Kakfa
使用方式
CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
...
) ENGINE = Kafka()
SETTINGS
kafka_broker_list = 'host:port',
kafka_topic_list = 'topic1,topic2,...',
kafka_group_name = 'group_name',
kafka_format = 'data_format'[,]
[kafka_row_delimiter = 'delimiter_symbol',]
[kafka_schema = '',]
[kafka_num_consumers = N,]
[kafka_max_block_size = 0,]
[kafka_skip_broken_messages = N,]
[kafka_commit_every_batch = 0,]
[kafka_thread_per_consumer = 0]
kafka_broker_list
:逗号分隔的brokers地址 (localhost:9092).kafka_topic_list
:Kafka 主题列表,多个主题用逗号分隔.kafka_group_name
:消费者组.kafka_format
– Message format. 比如JSONEachRow
、JSON、CSV等等
使用示例
在kafka中创建ck_topic主题,并向该主题写入数据
CREATE TABLE kafka_table (id UInt64,name String) ENGINE = Kafka() SETTINGS kafka_broker_list = 'hadoop00:9092',kafka_topic_list = 'test0930',kafka_group_name = 'group1',kafka_format = 'JSONEachRow';
-- 查询
select * from kafka_table ;
-- 添加数据
[root@hadoop01 ~]# kafka-console-producer.sh --broker-list hadoop01:9092 --topic test0930
>{"id":1,"name":"zs"}
>{"id":2,"name":"ls","age":16}
注意点
当我们一旦查询完毕之后,ClickHouse会删除表内的数据,其实Kafka表引擎只是一个数据管道,我们可以通过物化视图的方式访问Kafka中的数据。
- 首先创建一张Kafka表引擎的表,用于从Kafka中读取数据
- 然后再创建一张普通表引擎的表,比如MergeTree,面向终端用户使用
- 最后创建物化视图,用于将Kafka引擎表实时同步到终端用户所使用的表中
-- 创建Kafka引擎表
CREATE TABLE kafka_table_consumer (
id UInt64,
name String
) ENGINE = Kafka()
SETTINGS
kafka_broker_list = 'hadoop00:9092',
kafka_topic_list = 'test0930',
kafka_group_name = 'group2',
kafka_format = 'JSONEachRow'
;
-- 创建一张终端用户使用的表
CREATE TABLE kafka_table_mergetree (
id UInt64 ,
name String
)ENGINE=MergeTree()
ORDER BY id
;
-- 创建物化视图,同步数据
CREATE MATERIALIZED VIEW consumer TO kafka_table_mergetree
AS SELECT id,name FROM kafka_table_consumer;
-- 查询,多次查询,已经被查询的数据依然会被输出
select * from kafka_table_mergetree;
5.6 Memory表引擎
Memory表引擎直接将数据保存在内存中,数据既不会被压缩也不会被格式转换。当ClickHouse服务重启的时候,Memory表内的数据会全部丢失。一般在测试时使用。
CREATE TABLE table_memory(id UInt64,name String) ENGINE = Memory();
5.7 Distributed表引擎
使用方式
Distributed表引擎是分布式表的代名词,它自身不存储任何数据,数据都分散存储在某一个分片
上,能够自动路由数据至集群中的各个节点,所以Distributed表引擎需要和其他数据表引擎一起协同工作
。
所以,一张分布式表底层会对应多个本地分片数据表,由具体的分片表存储数据,分布式表与分片表是一对多的关系
Distributed表引擎的定义形式如下所示
Distributed(cluster_name, database_name, table_name[, sharding_key])
各个参数的含义分别如下:
- cluster_name:集群名称,与集群配置中的自定义名称相对应。
- database_name:数据库名称
- table_name:表名称
- sharding_key:可选的,用于分片的key值,在数据写入的过程中,分布式表会依据分片key的规则,将数据分布到各个节点的本地表。
注:
创建分布式表是读时检查的机制,也就是说对创建分布式表和本地表的顺序并没有强制要求。
同样值得注意的是,在上面的语句中使用了ON CLUSTER分布式DDL,这意味着在集群的每个分片节点上,都会创建一张Distributed表,这样便可以从其中任意一端发起对所有分片的读、写请求。
使用示例
参考第四章中的使用即可。
5.8 特别类型表引擎
Distributed 、MaterializedView(物化视图)、 Dictionary 、Merge 、File、Null 、Set 、Join 、 URL View、Memory 、 Buffer
第六章 数据类型
官网地址:https://clickhouse.tech/docs/zh/sql-reference/data-types/
AggregateFunction(name, types_of_arguments…)
聚合函数的中间状态,可以通过聚合函数名称加-State
后缀的形式得到它。与此同时,当您需要访问该类型的最终状态数据时,您需要以相同的聚合函数名加-Merge
后缀的形式来得到最终状态数据。
AggregateFunction
— 参数化的数据类型。
参数
-
聚合函数名
如果函数具备多个参数列表,请在此处指定其他参数列表中的值。
-
聚合函数参数的类型
UInt8,UInt16,UInt32,UInt64,Int8,Int16,Int32,Int64
固定长度的整型,包括有符号整型或无符号整型。
整型范围
- Int8-[-128:127]
- Int16-[-32768:32767]
- Int32-[-2147483648:2147483647]
- Int64-[-9223372036854775808:9223372036854775807]
无符号整型范围
- UInt8-[0:255]
- UInt16-[0:65535]
- UInt32-[0:4294967295]
- UInt64-[0:18446744073709551615]
允许用特殊标记 (NULL) 表示«缺失值»,可以与 TypeName
的正常值存放一起。例如,Nullable(Int8)
类型的列可以存储 Int8
类型值,而没有值的行将存储 NULL
。
对于 TypeName
,不能使用复合数据类型 阵列 和 元组。复合数据类型可以包含 Nullable
类型值,例如Array(Nullable(Int8))
。
Nullable
类型字段不能包含在表索引中。
除非在 ClickHouse 服务器配置中另有说明,否则 NULL
是任何 Nullable
类型的默认值。
没有单独的类型来存储布尔值。可以使用 UInt8 类型,取值限制为 0 或 1。
Nested(Name1 Type1, Name2 Type2, …)嵌套数据结构
第七章 函数
第八章 bitmap
ClickHouse Bitmap官网:
中文:https://clickhouse.tech/docs/zh/sql-reference/functions/bitmap-functions/
英文:https://clickhouse.tech/docs/en/sql-reference/functions/bitmap-functions/
相同机器,mysql的表映射到clickhouse,相同的查询sql,clickhous需要网络传输,小数据量简单查询表现clickhouse并不优秀