ClickHouse的表引擎介绍（三）

引入表引擎的概念以及特点
一、TinyLog
二、Memory
三、MergeTree
四、ReplacingMergeTree
五、SummingMergeTree
六、Integrations（集成引擎）
- - ODBC
- - JDBC
- - MySQL
- - MongoDB
- - HDFS
- - S3
- - Kafka
- - RabbitMQ
- - PostgreSQL

引入表引擎的概念以及特点

官方网站：https://clickhouse.com/docs/zh/engines/table-engines/

表引擎决定了如何存储表的数据。包括:

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

➢ 支持哪些查询以及如何支持

比如数组不能在merge引擎中使用。

➢ 并发数据访问。

➢ 索引的使用(如果存在)。

➢ 是否可以执行多线程请求。

➢ 数据复制参数。

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

特别注意：引擎的名称大小写敏感。

一、TinyLog

以列文件的形式保存在磁盘上，不支持索引，没有并发控制。一般保存少量数据的小表，生产环境上作用有限。可以用于平时练习测试用。

代码语言：javascript复制

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

二、Memory

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

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

三、MergeTree

ClickHouse 中最强大的表引擎当属 MergeTree(合并树)引擎及该系列(MergeTree) 中的其他引擎*，支持索引和分区。而且基于 MergeTree，还衍生除了很多小弟，也是非常有特色的引擎。 MergeTree 其实还有很多参数(绝大多数用默认值即可)，但是三个参数是更加重要的，也涉及了关于 MergeTree 的很多概念。

相当于Inoodb在MySQL中的地位

主要特点:

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

该类型的引擎：

MergeTree
ReplacingMergeTree
SummingMergeTree
AggregatingMergeTree
CollapsingMergeTree
VersionedCollapsingMergeTree
GraphiteMergeTree

3.1、简单的创建表和插入语句

建表语句

代码语言：javascript复制

create table t_order_mt(   id UInt32,   sku_id String,   total_amount Decimal(16,2),   create_time Datetime) engine =MergeTreepartition by toYYYYMMDD(create_time) primary key (id)order by (id,sku_id);

插入数据

代码语言：javascript复制

insert into t_order_mt values  (101,'sku_001',1000.00,'2020-06-01 12:00:00') , (102,'sku_002',2000.00,'2020-06-01 11:00:00'), (102,'sku_004',2500.00,'2020-06-01 12:00:00'),  (102,'sku_002',2000.00,'2020-06-01 13:00:00'),  (102,'sku_002',12000.00,'2020-06-01 13:00:00'),  (102,'sku_002',600.00,'2020-06-02 12:00:00');

3.2、重要的参数

一、partition by 分区(可选)

1）作用：分区的目的主要是降低扫描的范围，优化查询速度。如果不指明partition by的话，只会使用一个分区

2）分区目录：MergeTree 是以列文件索引文件表定义文件组成的，但是如果设定了分区那么这些文件就会保存到不同的分区目录中。

3）并行：分区后，面对涉及跨分区的查询统计，ClickHouse 会以分区为单位并行处理。

4）数据写入与分区合并：任何一个批次的数据写入都会产生一个临时分区，不会纳入任何一个已有的分区。写入后的某个时刻(大概 10-15 分钟后)，ClickHouse 会自动执行合并操作(等不及也可以手动。通过 optimize 执行)，把临时分区的数据，合并到已有分区中。

代码语言：javascript复制

optimize table xxxx final;

二、primary key 主键(可选)

ClickHouse 中的主键，和其他数据库不太一样，它只提供了数据的一级索引，但是却不是唯一约束。这就意味着是可以存在相同 primary key 的数据的。主键的设定主要依据是查询语句中的 where 条件。根据条件通过对主键进行某种形式的二分查找，能够定位到对应的index granularity,避免了全表扫描。 index granularity: 直接翻译的话就是索引粒度，指在稀疏索引中两个相邻索引对应数据的间隔。ClickHouse 中的 MergeTree 默认是 8192。官方不建议修改这个值，除非该列存在大量重复值，比如在一个分区中几万行才有一个不同数据。

稀疏索引:稀疏索引的好处就是可以用很少的索引数据，定位更多的数据，代价就是只能定位到索引粒度的第一行，然后再进行进行一点扫描。

三、order by(必选)

order by 设定了分区内的数据按照哪些字段顺序进行有序保存。 order by 是 MergeTree 中唯一一个必填项，甚至比 primary key 还重要，因为当用户不设置主键的情况，很多处理会依照 order by 的字段进行处理 要求:主键必须是 order by 字段的前缀字段。 比如 order by 字段是 (id,sku_id) 那么主键必须是 id 或者(id,sku_id)

四、二级索引

目前在 ClickHouse 的官网上二级索引的功能在 v20.1.2.4 之前是被标注为实验性的，在这个版本之后默认是开启的。

1)老版本使用二级索引前需要增加设置

是否允许使用实验性的二级索引(v20.1.2.4 开始，这个参数已被删除，默认开启) set allow_experimental_data_skipping_indices=1;

2)创建测试表

代码语言：javascript复制

create table t_order_mt2(        id UInt32,        sku_id String,        total_amount Decimal(16,2),        create_time Datetime,        INDEX a total_amount TYPE minmax GRANULARITY 5 ) engine =MergeTree    partition by toYYYYMMDD(create_time)    primary key (id)    order by (id, sku_id)    settings index_granulrity = 8192;

其中 GRANULARITY N 是设定二级索引对于一级索引粒度的粒度。

3)插入数据

代码语言：javascript复制

insert into t_order_mt values  (101,'sku_001',1000.00,'2020-06-01 12:00:00') , (102,'sku_002',2000.00,'2020-06-01 11:00:00'), (102,'sku_004',2500.00,'2020-06-01 12:00:00'),  (102,'sku_002',2000.00,'2020-06-01 13:00:00'),  (102,'sku_002',12000.00,'2020-06-01 13:00:00'),  (102,'sku_002',600.00,'2020-06-02 12:00:00');

五、数据TTL

TTL 即 Time To Live，MergeTree 提供了可以管理数据表或者列的生命周期的功能。

代码语言：javascript复制

total_amount Decimal(16,2) TTL create_time interval 10 SECOND, create_time Datetime

1)列级别 TTL

代码语言：javascript复制

create table t_order_mt3(        id UInt32,        sku_id String,        total_amount Decimal(16,2) TTL create_time interval 10 SECOND, create_time Datetime ) engine =MergeTree     partition by toYYYYMMDD(create_time)     primary key (id)     order by (id, sku_id);

代码语言：javascript复制

    到期后，指定的字段数据归0。

2)表级 TTL

下面的这条语句是数据会在 create_time 之后 10 秒丢失

代码语言：javascript复制

alter table t_order_mt3 MODIFY TTL create_time   INTERVAL 10 SECOND;

涉及判断的字段必须是 Date 或者 Datetime 类型，推荐使用分区的日期字段。

能够使用的时间周期:

SECOND
MINUTE
HOUR
DAY
WEEK
MONTH
QUARTER
YEAR

四、ReplacingMergeTree

ReplacingMergeTree 是 MergeTree 的一个变种，它存储特性完全继承 MergeTree，只是多了一个去重的功能。尽管 MergeTree 可以设置主键，但是 primary key 其实没有唯一约束的功能。如果你想处理掉重复的数据，可以借助这个 ReplacingMergeTree。

1)去重时机

数据的去重只会在合并的过程中出现。合并会在未知的时间在后台进行，所以你无法预先作出计划。有一些数据可能仍未被处理。

2)去重范围

如果表经过了分区，去重只会在分区内部进行去重，不能执行跨分区的去重。所以 ReplacingMergeTree 能力有限， ReplacingMergeTree 适用于在后台清除重复的数据以节省空间，但是它不保证没有重复的数据出现。

创建表

代码语言：javascript复制

create table t_order_rmt(   id UInt32,sku_id String,total_amount Decimal(16,2) , create_time Datetime) engine =ReplacingMergeTree(create_time)  partition by toYYYYMMDD(create_time)  primary key (id)  order by (id, sku_id);

ReplacingMergeTree() 填入的参数为版本字段，重复数据保留版本字段值最大的。 如果不填版本字段，默认按照插入顺序保留最后一条。

向表中插入数据

代码语言：javascript复制

insert into t_order_rmt values (101,'sku_001',1000.00,'2020-06-01 12:00:00') , (102,'sku_002',2000.00,'2020-06-01 11:00:00'), (102,'sku_004',2500.00,'2020-06-01 12:00:00'), (102,'sku_002',2000.00,'2020-06-01 13:00:00'), (102,'sku_002',12000.00,'2020-06-01 13:00:00'), (102,'sku_002',600.00,'2020-06-02 12:00:00');

执行第一次查询

代码语言：javascript复制

select * from t_order_rmt;

手动合并

代码语言：javascript复制

OPTIMIZE TABLE t_order_rmt FINAL;

再执行一次查询

代码语言：javascript复制

select * from t_order_rmt;

通过测试得到结论

➢ 实际上是使用 order by 字段作为唯一键 ➢ 去重不能跨分区 ➢ 只有同一批插入(新版本)或合并分区时才会进行去重 ➢ 认定重复的数据保留，版本字段值最大的 ➢ 如果版本字段相同则按插入顺序保留最后一笔

五、SummingMergeTree

对于不查询明细，只关心以维度进行汇总聚合结果的场景。 如果只使用普通的 MergeTree 的话，无论是存储空间的开销，还是查询时临时聚合的开销都比较大。ClickHouse 为了这种场景，提供了一种能够“预聚合”的引擎 SummingMergeTree

步骤如上，修改创建表时表引擎为SummingMergeTree

总结：

➢ 以SummingMergeTree()中指定的列作为汇总数据列
➢ 可以填写多列必须数字列，如果不填，以所有非维度列且为数字列的字段为汇总数据列
➢ 以 order by 的列为准，作为维度列
➢ 其他的列按插入顺序保留第一行
➢ 不在一个分区的数据不会被聚合
➢ 只有在同一批次插入(新版本)或分片合并时才会进行聚合

举例：

能不能直接执行以下 SQL 得到汇总值

代码语言：javascript复制

select total_amount from XXX where province_name=’’ and create_date=’xxx’

不行，可能会包含一些还没来得及聚合的临时明细

如果要是获取汇总值，还是需要使用 sum 进行聚合，这样效率会有一定的提高，但本身 ClickHouse 是列式存储的，效率提升有限，不会特别明显。

代码语言：javascript复制

select sum(total_amount) from province_name=’’ and create_date=‘xxx’

六、Integrations（集成引擎）

ClickHouse 提供了多种方式来与外部系统集成，包括表引擎。像所有其他的表引擎一样，使用CREATE TABLE或ALTER TABLE查询语句来完成配置。然后从用户的角度来看，配置的集成看起来像查询一个正常的表，但对它的查询是代理给外部系统的。这种透明的查询是这种方法相对于其他集成方法的主要优势之一，比如外部字典或表函数，它们需要在每次使用时使用自定义查询方法。

以下是支持的集成方式:

- ODBC

- JDBC

- MySQL

MySQL 引擎可以对存储在远程 MySQL 服务器上的数据执行 SELECT 查询。

调用格式：

代码语言：javascript复制

MySQL('host:port', 'database', 'table', 'user', 'password'[, replace_query, 'on_duplicate_clause']);

调用参数

host:port — MySQL 服务器地址。
database — 数据库的名称。
table — 表名称。
user — 数据库用户。
password — 用户密码。
replace_query — 将 INSERT INTO 查询是否替换为 REPLACE INTO 的标志。如果 replace_query=1，则替换查询
'on_duplicate_clause' — 将 ON DUPLICATE KEY UPDATE 'on_duplicate_clause' 表达式添加到 INSERT 查询语句中。例如：impression = VALUES(impression) impression。如果需要指定 'on_duplicate_clause'，则需要设置 replace_query=0。如果同时设置 replace_query = 1 和 'on_duplicate_clause'，则会抛出异常。

此时，简单的 WHERE 子句（例如 =, !=, >, >=, <, <=）是在 MySQL 服务器上执行。

其余条件以及 LIMIT 采样约束语句仅在对MySQL的查询完成后才在ClickHouse中执行。

MySQL 引擎不支持可为空数据类型，因此，当从MySQL表中读取数据时，NULL 将转换为指定列类型的默认值（通常为0或空字符串）。

- MongoDB

- HDFS

- S3

- Kafka

此引擎与 Apache Kafka 结合使用。

Kafka 特性：

发布或者订阅数据流。
容错存储机制。
处理流数据。

老版格式：

代码语言：javascript复制

Kafka (    kafka_broker_list,    kafka_topic_list,     kafka_group_name,    kafka_format    [        kafka_row_delimiter,        kafka_schema,         kafka_num_consumers    ])

新版格式：

代码语言：javascript复制

Kafka SETTINGSkafka_broker_list = 'localhost:9092',kafka_topic_list = 'topic1,topic2',kafka_group_name = 'group1',kafka_format = 'JSONEachRow',kafka_row_delimiter = 'n',kafka_schema = '',kafka_num_consumers = 2

必要参数：

kafka_broker_list – 以逗号分隔的 brokers 列表 (localhost:9092)。
kafka_topic_list – topic 列表 (my_topic)。
kafka_group_name – Kafka 消费组名称 (group1)。如果不希望消息在集群中重复，请在每个分片中使用相同的组名。
kafka_format – 消息体格式。使用与 SQL 部分的 FORMAT 函数相同表示方法，例如 JSONEachRow。了解详细信息，请参考 Formats 部分。

可选参数：

kafka_row_delimiter - 每个消息体（记录）之间的分隔符。
kafka_schema – 如果解析格式需要一个 schema 时，此参数必填。例如，普罗托船长需要 schema 文件路径以及根对象 schema.capnp:Message 的名字。
kafka_num_consumers – 单个表的消费者数量。默认值是：1，如果一个消费者的吞吐量不足，则指定更多的消费者。消费者的总数不应该超过 topic 中分区的数量，因为每个分区只能分配一个消费者。

示例：

代码语言：javascript复制

`CREATE TABLE queue (    timestamp UInt64,    level String,    message String) ENGINE = Kafka('localhost:9092', 'topic', 'group1', 'JSONEachRow');SELECT * FROM queue LIMIT 5;CREATE TABLE dmp_log.sync_kafka_u0(`json` Nullable(String))ENGINE = KafkaSETTINGS kafka_broker_list = '172.30.xxx.xx:9092',kafka_topic_list = 'ck.t0',kafka_group_name = 't0.g1',kafka_format = 'JSONAsString',kafka_row_delimiter = 'n',kafka_skip_broken_messages = 1,kafka_num_consumers = 1;CREATE TABLE queue2 (    timestamp UInt64,    level String,    message String) ENGINE = Kafka('localhost:9092', 'topic', 'group1')SETTINGS kafka_format = 'JSONEachRow',kafka_num_consumers = 4;`

消费的消息会被自动追踪，因此每个消息在不同的消费组里只会记录一次。如果希望获得两次数据，则使用另一个组名创建副本。

消费组可以灵活配置并且在集群之间同步。例如，如果群集中有10个主题和5个表副本，则每个副本将获得2个主题。如果副本数量发生变化，主题将自动在副本中重新分配。了解更多信息请访问 http://kafka.apache.org/intro。

SELECT 查询对于读取消息并不是很有用（调试除外），因为每条消息只能被读取一次。使用物化视图创建实时线程更实用。您可以这样做：

使用引擎创建一个 Kafka 消费者并作为一条数据流。
创建一个结构表。
创建物化视图，改视图会在后台转换引擎中的数据并将其放入之前创建的表中。

当 MATERIALIZED VIEW 添加至引擎，它将会在后台收集数据。可以持续不断地从 Kafka 收集数据并通过 SELECT 将数据转换为所需要的格式。

示例：

代码语言：javascript复制

CREATE MATERIALIZED VIEW dmp_log.view_log_t0 TO dmp_log.ods_log_t0(`data_type` Int16,`org_id` Int64,`org_name` String,`product_key` String,`device_key` String,`properties` String) ASSELECTJSONExtractInt(json,'dataType') data_type,JSONExtractFloat(json,'orgId') org_id,JSONExtractString(json,'orgName') org_name,JSONExtractString(json,'productKey') product_key,JSONExtractString(json,'deviceKey') device_key,JSONExtractString(json,'properties') propertiesFROM dmp_log.sync_kafka_u0 where _topic='ck.t0';

为了提高性能，接受的消息被分组为 max_insert_block_size 大小的块。如果未在 stream_flush_interval_ms 毫秒内形成块，则不关心块的完整性，都会将数据刷新到表中。

停止接收主题数据或更改转换逻辑，请 detach 物化视图：

DETACH TABLE consumer; ATTACH TABLE consumer;

如果使用 ALTER 更改目标表，为了避免目标表与视图中的数据之间存在差异，推荐停止物化视图。

配置

与 GraphiteMergeTree 类似，Kafka 引擎支持使用ClickHouse配置文件进行扩展配置。可以使用两个配置键：全局 (kafka) 和主题级别 (kafka_*)。首先应用全局配置，然后应用主题级配置（如果存在）。

代码语言：javascript复制

`<!-- Global configuration options for all tables of Kafka engine type --><kafka><debug>cgrp</debug><auto_offset_reset>smallest</auto_offset_reset></kafka><!-- Configuration specific for topic "logs" --><kafka_logs><retry_backoff_ms>250</retry_backoff_ms><fetch_min_bytes>100000</fetch_min_bytes></kafka_logs>`

- RabbitMQ

该引擎允许 ClickHouse 与 RabbitMQ 进行集成.

RabbitMQ 可以让你:

发布或订阅数据流。
在数据流可用时进行处理。

创建一张表

代码语言：javascript复制

`CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster](name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],...) ENGINE = RabbitMQ SETTINGSrabbitmq_host_port = 'host:port',rabbitmq_exchange_name = 'exchange_name',rabbitmq_format = 'data_format'[,][rabbitmq_exchange_type = 'exchange_type',][rabbitmq_routing_key_list = 'key1,key2,...',][rabbitmq_row_delimiter = 'delimiter_symbol',][rabbitmq_schema = '',][rabbitmq_num_consumers = N,][rabbitmq_num_queues = N,][rabbitmq_queue_base = 'queue',][rabbitmq_deadletter_exchange = 'dl-exchange',][rabbitmq_persistent = 0,][rabbitmq_skip_broken_messages = N,][rabbitmq_max_block_size = N,][rabbitmq_flush_interval_ms = N]`

必要参数:

rabbitmq_host_port – 主机名:端口号 (比如, localhost:5672).
rabbitmq_exchange_name – RabbitMQ exchange 名称.
rabbitmq_format – 消息格式. 使用与SQLFORMAT函数相同的标记，如JSONEachRow。更多信息，请参阅 Formats 部分.

可选参数:

rabbitmq_exchange_type – RabbitMQ exchange 的类型: direct, fanout, topic, headers, consistent_hash. 默认是: fanout.
rabbitmq_routing_key_list – 一个以逗号分隔的路由键列表.
rabbitmq_row_delimiter – 用于消息结束的分隔符.
rabbitmq_schema – 如果格式需要模式定义，必须使用该参数。比如, Cap’n Proto 需要模式文件的路径以及根 schema.capnp:Message 对象的名称.
rabbitmq_num_consumers – 每个表的消费者数量。默认：1。如果一个消费者的吞吐量不够，可以指定更多的消费者.
rabbitmq_num_queues – 队列的总数。默认值: 1. 增加这个数字可以显著提高性能.
rabbitmq_queue_base - 指定一个队列名称的提示。这个设置的使用情况如下.
rabbitmq_deadletter_exchange - 为dead letter exchange指定名称。你可以用这个 exchange 的名称创建另一个表，并在消息被重新发布到 dead letter exchange 的情况下收集它们。默认情况下，没有指定 dead letter exchange。Specify name for a dead letter exchange.
rabbitmq_persistent - 如果设置为 1 (true), 在插入查询中交付模式将被设置为 2 (将消息标记为 ‡persistent’). 默认是: 0.
rabbitmq_skip_broken_messages – RabbitMQ 消息解析器对每块模式不兼容消息的容忍度。默认值：0. 如果 rabbitmq_skip_broken_messages = N，那么引擎将跳过 N 个无法解析的 RabbitMQ 消息（一条消息等于一行数据）。
rabbitmq_max_block_size
rabbitmq_flush_interval_ms

同时，格式的设置也可以与 rabbitmq 相关的设置一起添加。

示例:

代码语言：javascript复制

`CREATE TABLE queue (    key UInt64,    value UInt64,    date DateTime) ENGINE = RabbitMQ SETTINGS rabbitmq_host_port = 'localhost:5672',rabbitmq_exchange_name = 'exchange1',rabbitmq_format = 'JSONEachRow',rabbitmq_num_consumers = 5,date_time_input_format = 'best_effort';`

RabbitMQ 服务器配置应使用 ClickHouse 配置文件添加。

必要配置:

代码语言：javascript复制

`<rabbitmq><username>root</username><password>clickhouse</password></rabbitmq>`

可选配置:

代码语言：javascript复制

`<rabbitmq><vhost>clickhouse</vhost></rabbitmq>`

描述

SELECT对于读取消息不是特别有用（除了调试），因为每个消息只能读取一次。使用物化视图创建实时线程更为实用。要做到这一点:

使用引擎创建一个 RabbitMQ 消费者，并将其视为一个数据流。
创建一个具有所需结构的表。
创建一个物化视图，转换来自引擎的数据并将其放入先前创建的表中。

当物化视图加入引擎时，它开始在后台收集数据。这允许您持续接收来自 RabbitMQ 的消息，并使用 SELECT 将它们转换为所需格式。一个 RabbitMQ 表可以有多个你需要的物化视图。

数据可以根据rabbitmq_exchange_type和指定的rabbitmq_routing_key_list进行通道。每个表不能有多于一个 exchange。一个 exchange 可以在多个表之间共享 - 因为可以使用路由让数据同时进入多个表。

Exchange 类型的选项:

direct - 路由是基于精确匹配的键。例如表的键列表: key1,key2,key3,key4,key5, 消息键可以是等同他们中的任意一个.
fanout - 路由到所有的表 (exchange 名称相同的情况) 无论是什么键都是这样.
topic - 路由是基于带有点分隔键的模式. 比如: .logs, records.*.*.2020, .2018,*.2019,*.2020.
headers - 路由是基于key=value的匹配，设置为x-match=all或x-match=any. 例如表的键列表: x-match=all,format=logs,type=report,year=2020.
consistent_hash - 数据在所有绑定的表之间均匀分布 (exchange 名称相同的情况). 请注意，这种 exchange 类型必须启用 RabbitMQ 插件: rabbitmq-plugins enable rabbitmq_consistent_hash_exchange.

设置rabbitmq_queue_base可用于以下情况:

来让不同的表共享队列, 这样就可以为同一个队列注册多个消费者，这使得性能更好。如果使用rabbitmq_num_consumers和/或rabbitmq_num_queues设置，在这些参数相同的情况下，实现队列的精确匹配。
以便在不是所有消息都被成功消费时，能够恢复从某些持久队列的阅读。要从一个特定的队列恢复消耗 - 在rabbitmq_queue_base设置中设置其名称，不要指定rabbitmq_num_consumers和rabbitmq_num_queues（默认为1）。要恢复所有队列的消费，这些队列是为一个特定的表所声明的 - 只要指定相同的设置。rabbitmq_queue_base, rabbitmq_num_consumers, rabbitmq_num_queues。默认情况下，队列名称对表来说是唯一的。
以重复使用队列，因为它们被声明为持久的，并且不会自动删除。可以通过任何 RabbitMQ CLI 工具删除）

为了提高性能，收到的消息被分组为大小为 max_insert_block_size 的块。如果在stream_flush_interval_ms毫秒内没有形成数据块，无论数据块是否完整，数据都会被刷到表中。

如果rabbitmq_num_consumers和/或rabbitmq_num_queues设置与rabbitmq_exchange_type一起被指定，那么:

必须启用rabbitmq-consistent-hash-exchange 插件.
必须指定已发布信息的 message_id属性（对于每个信息/批次都是唯一的）。

对于插入查询时有消息元数据，消息元数据被添加到每个发布的消息中:messageID和republished标志（如果值为true，则表示消息发布不止一次） - 可以通过消息头访问。

不要在插入和物化视图中使用同一个表。

示例:

代码语言：javascript复制

`CREATE TABLE queue (    key UInt64,    value UInt64) ENGINE = RabbitMQ SETTINGS rabbitmq_host_port = 'localhost:5672',rabbitmq_exchange_name = 'exchange1',rabbitmq_exchange_type = 'headers',rabbitmq_routing_key_list = 'format=logs,type=report,year=2020',rabbitmq_format = 'JSONEachRow',rabbitmq_num_consumers = 5;CREATE TABLE daily (key UInt64, value UInt64)ENGINE = MergeTree() ORDER BY key;CREATE MATERIALIZED VIEW consumer TO dailyAS SELECT key, value FROM queue;SELECT key, value FROM daily ORDER BY key;`

虚拟列

_exchange_name - RabbitMQ exchange 名称.
_channel_id - 接收消息的消费者所声明的频道ID.
_delivery_tag - 收到消息的DeliveryTag. 以每个频道为范围.
_redelivered - 消息的redelivered标志.
_message_id - 收到的消息的ID；如果在消息发布时被设置，则为非空.
_timestamp - 收到的消息的时间戳；如果在消息发布时被设置，则为非空.- PostgreSQL

更多的引擎请移驾ClickHouse官网：表引擎https://clickhouse.com/docs/zh/engines/table-engines/

rabbitmq 编程算法数据库 sql 云数据库SQLServer

0 人点赞