[ClickHouse]表引擎学习分享

作者: 愚公300代 | 来源:发表于2021-07-04 19:45 被阅读0次

设计理念

Everything is table（万物皆表），数据表就是ClickHouse和外部交互的接口。在数据表背后无论连接的是本地文件、HDFS、Zookeeper还是其它服务，终端用户始终只需要面对数据表，只需要用SQL查询语言。
处理各种数据场景走内部集成的路线，既与外部系统（其它数据库，消息中间件或者是服务接口）的集成直接在数据库内部实现。
面向表编程，一张数据表最终能够提供哪些功能、拥有哪些特性、数据会议什么格式被保存以及数据会怎样被加载，这些都将由它的表引擎决定。

表引擎特性

表引擎	概念	引擎
合并树类型	MergeTree提供主键索引，数据分区，数据副本和数据采样等基本能力，其它表引擎在MergeTree上各有所长；ReplacingMergeTree具有删除重复数据的特性；SummingMergeTree会按照排序键自动聚合数据；Replicated系列支持数据副本	MergeTree,SummingMergeTree ...
外部存储类型	外部存储表引擎直接从其他的存储系统读取数据，例如直接读取HDFS的文件或者MySQL数据库的表。这些引擎只负责元数据管理和数据查询，而它们自身通常并不负责数据的写入，数据文件直接由外部系统提供。	HDFS,MySQL,JDBC,Kafka,File
内存类型	面向内存查询，数据从内存中直接访问。除了Memory引擎之外，其它几款都会将数据写入磁盘以防数据丢失。在数据被加载时，数据会被全部加载到内存、以供查询使用。全部加载到内存：意味着有更好的查询性能；但是如果加载的数据量过大，就会有极大的内存消耗和负担	Memory,Set,Join,Buffer,
日志类型	数据量很小（100万以下），同时查询场景也比较简单，并且是“一次”写入多次查询方式，可以使用日志类型；共性：不支持索引和分区等高级特性；不支持并发读写，当针对一张日志写入数据时，针对这张表的查询会被阻塞，直至写入动作结束；拥有物理存储，数据会被保存到本地文件中	TinyLog,StripeLog,Log
接口类型	本身不保存数据，而是像粘合剂一样整合其它数据表；使用时，不用关心底层复杂性，像接口一样为用户提供统一访问页面	Merge,Dictionary,Distribute
其它类型	扩充ClickHouse的能力边界	Live View,NULL,URL

引擎使用场景

表引擎使用

SummingMergeTree（MergeTree家族）

使用

数据聚合

聚合前数据

聚合前数据
聚合后数据

聚合后数据

使用说明

PRIMARY KEY可与ORDER BY不同，PRIMARY KEY作为主键索引，ORDER BY作为聚合条件。
PRIMARY KEY是ORDER BY的前缀，索引和数据仍然具有对应关系；因为数据以ORDER BY排序，索引以PRIMARY KEY排序，PRIMARY KEY是ORDER BY的前缀，那么索引也是ORDER BY有序的，同一个排序标准，产生相同的数据顺序，所以，索引和数据仍然具有对应关系。
ORDER BY可以可以减少，GROUP BY(A, B, C, D) --> GROUP BY(A, B); ORDER BY可以增加新的列。

工作中使用

SummingMergeTree聚合ClickHouse Shard中数据，同时通过Distribute表向外提供服务
PARTITION BY toStartOfHour(event_date) + PRIMARY KEY (event_date, network_id)
ORDER BY (event_date, network_id, a, c, d) ...
TTL event_date + INTERVAL 35 DAY DELETE
SETTINGS index_granularity = 8192, replicated_deduplication_window = 0

原理

用ORDER BY排序键作为聚合数据的条件Key
只有在合并分区的时候才会触发汇总的逻辑
以数据分区为单位来聚合数据。当分区合并时，同一个数据分区内聚合Key相同的数据会被合并汇总，而不同分区之间的数据不会被汇总
如果在定义引擎时指定了columns汇总列（非主键的数值类型字段），则SUM汇总这些列字段；如果未指定，则聚合所有非主键的数值类型字段
在进行数据汇总时，因为分区内的数据已经基于ORDER BY排序，所以能够找到相邻且拥有相同聚合Key的数据
在汇总数据时，同一个分区内，相同聚合Key的多行数据会合并成一行。其中，汇总字段进行SUM计算；对于那些非汇总地段，则会使用第一行数据的取值。
支持嵌套结构，但列字段名称必须以Map后缀结尾。

Kafka（外部存储类型）

使用

kafka环境准备

Kafka docker 环境搭建

生产消息

数据data.json：

{ "id": "A001", "city": "wuhan", "v1": 10, "v2": 20, "create_time": "2019-08-10 17:00:00" }
{ "id": "A001", "city": "wuhan", "v1": 20, "v2": 30, "create_time": "2019-08-20 17:00:00" }
{ "id": "A001", "city": "zhuhai", "v1": 20, "v2": 30, "create_time": "2019-08-10 17:00:00" }
{ "id": "A001", "city": "wuhan", "v1": 10, "v2": 20, "create_time": "2019-02-10 09:00:00" }
{ "id": "A002", "city": "wuhan", "v1": 60, "v2": 50, "create_time": "2019-10-10 17:00:00" }

生产

kafka-console-producer --topic test --bootstrap-server localhost:9092 < data.json

创建kafka表

CREATE TABLE test_kafka
(
    `id` String,
    `city` String,
    `v1` UInt32,
    `v2` Float64,
    `create_time` DateTime
)
ENGINE = Kafka()
SETTINGS kafka_broker_list = '172.18.0.3:9092', kafka_topic_list = 'test', kafka_group_name = 'test', kafka_format = 'JSONEachRow', kafka_skip_broken_messages = 100

消费消息

SELECT * FROM test_kafka

查看kafka消费日志

cat /var/log/clickhouse-server/clickhouse-server.log | grep kafka

2021.06.30 16:23:03.749646 [ 75 ] {af851164-e7ce-46cd-be01-64f10ddec924} <Debug> executeQuery: (from 127.0.0.1:34022) select * from default.test_kafka;
2021.06.30 16:23:03.750176 [ 75 ] {af851164-e7ce-46cd-be01-64f10ddec924} <Debug> StorageKafka (test_kafka): Starting reading 1 streams
2021.06.30 16:23:07.262678 [ 91 ] {af851164-e7ce-46cd-be01-64f10ddec924} <Trace> StorageKafka (test_kafka): Polled batch of 98 messages. Offsets position: [ test[0:98] ]
2021.06.30 16:23:07.282639 [ 91 ] {af851164-e7ce-46cd-be01-64f10ddec924} <Warning> StorageKafka (test_kafka): Parsing of message (topic: test, partition: 0, offset: 77) return no rows.
2021.06.30 16:23:07.285524 [ 91 ] {af851164-e7ce-46cd-be01-64f10ddec924} <Trace> StorageKafka (test_kafka): Polled offset 98 (topic: test, partition: 0)
2021.06.30 16:23:07.339373 [ 91 ] {af851164-e7ce-46cd-be01-64f10ddec924} <Trace> StorageKafka (test_kafka): Committed offset 98 (topic: test, partition: 0)

原理

只负责元数据的管理和数据查询，不存储数据（外部存储引擎共性），支持从kafa消费消息，也可以向kafka中插入数据（Demo）
默认情况下，Kafka表引擎每隔500毫秒会拉取一次数据，时间由stream_poll_timeout_ms参数控制，数据首先会被放入缓存，在时机成熟时，缓存数据会被刷新到数据表
满足下列条件之一，触发刷新动作：
a. 当完成一个数据块儿的写入（数据块儿的大小由kafka_max_block_size参数控制，默认情况下65536)
b. 等待时间超过7500毫秒（stream_flush_interval_ms参数控制，默认7500ms）

Join（内存类型）

使用

CREATE TABLE join_tb1 (
    id UInt8,
    name String,
    time DateTime
) ENGINE = Log

INSERT INTO TABLE join_tb1 VALUES(1, 'ClickHouse', '2019-05-01 12:00:00'), 
(2, 'Spark', '2019-05-01 12:30:00'), (3, 'ElasticSearch', '2019-05-01 13:00:00');


CREATE TABLE id_join_tb1 (
    id UInt8,
    price UInt32,
    time DateTime
) ENGINE = Join(ANY, LEFT, id)

INSERT INTO TABLE id_join_tb1 VALUES (1, 100, '2019-05-01 11:55:00'),
(1, 105, '2019-05-01 11:10:00'),
(2, 90, '2019-05-01 12:01:00'),
(3, 80, '2019-05-01 11:55:00'),
(5, 70, '2019-05-01 11:55:00'),
(6, 60, '2019-05-01 11:55:00');

SELECT id, name, price FROM join_tb1 LEFT JOIN id_join_tb1 USING(id);

SELECT joinGet('id_join_tb1', 'price', toUInt8(1));

原理

ENGINE = Join(join_strictness, join_type, key1[, key2, ...])
join_strictness: 连接的精度，它决定了JOIN查询在连接数据时所使用的策略，目前支持ALL，ANY和ASOF三种类型
join_type: 连接左右两个数据集合的策略；交集，并集，笛卡尔积或其他形式，目前支持INNER，OUTER和CROSS；当join_type类型为ANY时，在数据写入时，join_key重复的数据会被自动忽略
join_key: 连接键，决定使用哪个列字段进行关联

Merge（接口类型）

使用

# 数据以年分表，使用Merge引擎进行粘合
CREATE TABLE test_table_2018(
    id String,
    create_time DateTime,
    code String
)ENGINE = MergeTree()
PARTITION BY toYYYYMM(create_time)
ORDER BY id

CREATE TABLE test_table_2019(
    id String,
    create_time DateTime,
    code String
)ENGINE = Log()
PARTITION BY toYYYYMM(create_time)
ORDER BY id

CREATE TABLE test_table_all as test_table_2018
ENGINE = MergeTree(currentDatebase(), '^test_table_')

原理

不存储数据，而是像粘合剂一样可以整合其他的数据表
被代理查询的数据表要在同一个数据库内，且拥有相同的表结构，但是它们可以使用不同的表引擎以及不同的分区定义（对于MergeTree而言）

URL（其它类型）

使用

/* GET users listing. */
router.get('/users', func (req, res, next)  {
    var result = ''
    for(let i=0; i<5; i++){
        result += '{"name":"nauu'+i+'"}\n';
    }
    res.send(result)    
})

/* POST user. */
router.post('/users', func (req, res)  {
    res.sendStatus(200) 
})

CREATE TABLE url_table (
    name String
)
ENGINE = URL('http://localhost:9688/users', JSONEachRow)

SELECT * FROM url_table

INSERT INTO TABLE url_table VALUES('nauu-insert')

原理

URL表引擎等价于HTTP客户端，它可以通过HTTP/HTTPS协议，直接访问远端的REST服务。
SELECT查询会被底层转换为GET请求
INSERT查询会被转换为POST请求

综合使用例子

Kafka + MATERIALIZED VIEW + ReplicateSummingMergeTree + Distributed
Kafka Engine Table：外部存储表，消费kafka消息
Materialize View: 当数据插入到kafka表时，执行select语句将数据进行transform后，插入到To表
SummingMergeTree: MergeTree家族表，支持partition summing，主键索引，数据分区，replica和数据采样；
Distribute 表：进行数据粘合，为用户提供统一的数据视图

引用

Docker HDFS
Docker HDFS 2
apt install netcat
Docker中容器之间通讯方式
 安装ifconfig apt install net-tools
查看docker容器ip地址 docker inspect kafka-docker_clickhouse-server_1 | grep IP
Kafka引擎
 ClickHouse原理解析与应用实践

网友评论

本文标题：[ClickHouse]表引擎学习分享

本文链接：https://www.haomeiwen.com/subject/kolvultx.html

延伸阅读

深度阅读

您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！