ClickHouse 表引擎介绍
发布于 2 个月前 作者 theseusyang 273 次浏览 来自 文档翻译

表引擎介绍:

ClickHouse表引擎决定了如下几个方面:

怎样存储数据 -将数据写到哪里, 怎样读取数据.

支持何种查询以及怎样支持.

并发数据访问.

索引的使用.

是否多线程的请求执行是可以的.

数据如何同步.

当读取数据时, 引擎只需要抽取必要的列簇. 然而,在一些场景下,查询可能在表引擎中是半处理状态.

在大多数场景中, 我们所使用的引擎主要是 MergeTree 家族.

AggregatingMergeTree

Buffer

CollapsingMergeTree

Distributed

External data for query processing

File(InputFormat)

Join

Kafka

Log

MaterializedView

Memory

Merge

Virtual columns

MergeTree

Null

ReplacingMergeTree

Data replication

ReplicatedMergeTree

ReplicatedCollapsingMergeTree

ReplicatedAggregatingMergeTree

ReplicatedSummingMergeTree

Creating replicated tables

Recovery after failures

Recovery after complete data loss

Converting from MergeTree to ReplicatedMergeTree

Converting from ReplicatedMergeTree to MergeTree

Recovery when metadata in the ZooKeeper cluster is lost or damaged

Resharding

Set

SummingMergeTree

TinyLog

View

Kafka

表引擎的后台为Kafka,Kafka 作为流数据平台具备三中关键能力:

它让你发布和订阅记录数据流。它类似消息队列或者企业级消息总线。

它让你以容错的方式来存储记录数据流。

它让你以流式的方式处理记录数据流。

Kafka(broker_list, topic_list, group_name, format[, schema])

引擎参数:

broker_list - 一个 brokers 列表((localhost:9092).

topic_list - Kafka队列列表用于消费(my_topic).

group_name -Kafka 消费者群组名称(group1).

每个消费者群组的偏移量被追踪,如果你想要跨集群一次性消费消息,你应该使用相同的群组名称

format - 格式的名称,用于反序列化消息. 它接受了相同的值作为 FORMATSQL 语句, 例如 JSONEachRow.

schema - 可选的 schema 值,需要一个 schema 来中断所消费的消息, 例如 Cap’n Proto 格式需要一个 schema 文件的路径和根对象(root object) -schema.capnp:Message. 自描述格式, 例如 JSON 不需要任何的 Schema.

例如:

CREATE TABLE queue (timestamp UInt64, level String, message String) ENGINE = Kafka(‘localhost:9092’, ‘topic’, ‘group1’, ‘JSONEachRow’);

SELECT * FROM queue LIMIT 5 已消费的消息自动在后台被追踪,这样每个消息都能够在一个消费者组中被读一次。如果你想要二次消费相同的消息,你能够创建一个带有不同组名的表的拷贝。消费者组是弹性的,跨集群同步的,因此,如果你有10个话题/分区(topic/partitions)和5个表的实例(instances),它将为每个实例自动分配2个话题/分区(topic/partitions)。如果你卸载一个表或者添加新的实例,它将自动重新平衡话题/分区(topic/partitions)分配。详情查看http://kafka.apache.org/intro。

然而直接读取消息并不是太有用,表引擎被用于构建实时摄取Pipeline, 使用MATERIALIZED VIEW。如果一个MATERIALIZED VIEW被挂载到一个 Kafka 表引擎,它将在后台进程中开始消费消息,推送到挂载的视图。它允许你从 Kafka 持续摄取消息,同时使用SELECT语句来转换这些数据到合适的格式。

示例:

CREATE TABLE queue (timestamp UInt64, level String, message String) ENGINE = Kafka(‘localhost:9092’, ‘topic’, ‘group1’, ‘JSONEachRow’);

CREATE MATERIALIZED VIEW daily ENGINE = SummingMergeTree(day, (day, level), 8192) AS

SELECT toDate(toDateTime(timestamp)) AS day, level, count() as total FROM queue GROUP BY day, level;

SELECT level, sum(total) FROM daily GROUP BY level; 当INSERT插入数据时,消息立即流入到挂载的视图中。为了提升处理性能,已消费的消息以批量的方式进行压缩 - max_insert_block_size。如果此批量消息不能被完成,刷新周期为 stream_flush_interval_ms (默认为7500ms),它将刷新来保证插入时间间隔。

Buffer

Buffers 将数据写入到内存中,周期性刷新数据到另外的表中。在读取操作的过程中,数据从 Buffer 和另外的表中同时读取。

Buffer(database, table, num_layers, min_time, max_time, min_rows, max_rows, min_bytes, max_bytes) 引擎参数:database, table - 表刷新数据。与数据库名称不同,你能够使用常量表达式返回字符串。

num_layers - 并行水平。

物理上看,表能够用独立 buffers 的‘num_layers’代表。默认值为16,min_time, max_time, min_rows, max_rows, min_bytes,同时max_bytes是从 buffer 刷新数据的条件。

数据从buffer 中刷新,写入到目标表,如果所有的‘min’条件或者最少一个‘max’条件组合时。

min_time, max_time - 从第一次写入到 buffer 的时间。

min_rows, max_rows - 在 buffer 中的行数。

min_bytes, max_bytes - 在 buffer 中字节的数量。

在写入操作的过程中,数据写入到一个‘num_layers’数的随机 buffers 中。或者,如果插入的数据部分足够大(大于 ‘max_rows’ 或者 ‘max_bytes’),它被直接写入到目标表,忽略此 buffer。

刷新数据被单独计算为每个‘num_layers’buffers。例如,如果num_layers = 16 and max_bytes = 100000000,最大的内存消耗是1.6 GB。

例如:

CREATE TABLE merge.hits_buffer AS merge.hits ENGINE = Buffer(merge, hits, 16, 10, 100, 10000, 1000000, 10000000, 100000000)

1 回复

这文档排版表示很难看啊

回到顶部