Skip to content
ClickHouse Docs
ClickHouse DocsClickHouse Docs
49.6kGet Started

AzureQueue 表引擎

View as MarkdownEdit this pageReport an issueSet up docs MCP

该引擎提供与 Azure Blob 存储 生态的集成，可实现流式数据导入。

创建表

CREATE TABLE test (name String, value UInt32)
    ENGINE = AzureQueue(...)
    [SETTINGS]
    [mode = '',]
    [after_processing = 'keep',]
    [keeper_path = '',]
    ...

引擎参数

AzureQueue 的参数与 AzureBlobStorage 表引擎支持的参数相同。请参阅此处的参数部分。

与 AzureBlobStorage 表引擎类似，用户可以使用 Azurite 模拟器在本地进行 Azure Storage 开发。更多详情请参阅此处。

示例

CREATE TABLE azure_queue_engine_table
(
    `key` UInt64,
    `data` String
)
ENGINE = AzureQueue('DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://azurite1:10000/devstoreaccount1/;', 'testcontainer', '*', 'CSV')
SETTINGS mode = 'unordered'

设置

支持的设置基本与 S3Queue 表引擎相同，只是不带 s3queue_ 前缀。请参阅完整设置列表。 如需查看为该表配置的设置列表，请使用 system.azure_queue_settings 表。该功能自 24.10 起可用。

以下设置仅适用于 AzureQueue，不适用于 S3Queue。

after_processing_move_connection_string

如果目标端是另一个 Azure 容器，则用于将已成功处理的文件移动到该容器的 Azure Blob 存储连接字符串。

可能的值：

  • String。

默认值：空字符串。

after_processing_move_container

如果目标端是另一个 Azure 容器，则此参数指定成功处理后的文件要移动到的容器名称。

可能的值：

  • String。

默认值：空字符串。

示例：

CREATE TABLE azure_queue_engine_table
(
    `key` UInt64,
    `data` String
)
ENGINE = AzureQueue('DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://azurite1:10000/devstoreaccount1/;', 'testcontainer', '*', 'CSV')
SETTINGS
    mode = 'unordered',
    after_processing = 'move',
    after_processing_move_connection_string = 'DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://azurite1:10000/devstoreaccount1/;',
    after_processing_move_container = 'dst-container';

对 AzureQueue 表引擎执行 SELECT

默认情况下，AzureQueue 表禁止执行 SELECT 查询。这遵循了常见的队列模式：数据读取一次后，就会从队列中移除。禁用 SELECT 是为了防止意外的数据丢失。 不过，在某些场景下，这样做也可能很有用。要启用此功能，需要将设置 stream_like_engine_allow_direct_select 设为 True。 AzureQueue 引擎还为 SELECT 查询提供了一个特殊设置：commit_on_select。将其设为 False 可在读取后保留队列中的数据，设为 True 则会将其移除。

描述

SELECT 对流式导入并不是特别有用 (调试除外) ，因为每个文件只能导入一次。更实用的做法是使用 materialized views 创建实时处理链路。为此：

  1. 使用该引擎创建一个表，用于从 Azure Blob 存储中的指定路径消费数据，并将其视为数据流。
  2. 创建一个具有所需结构的表。
  3. 创建一个 materialized view，将来自该引擎的数据转换后写入前面创建的表中。

当 MATERIALIZED VIEW 关联到该引擎时，它就会开始在后台收集数据。

该引擎的参数形式为 AzureQueue(connection_string, container_name, blobpath, format[, compression])。

示例：

CREATE TABLE azure_queue_engine_table (key UInt64, data String)
ENGINE=AzureQueue('DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://azurite1:10000/devstoreaccount1/;', 'testcontainer', '*', 'CSV')
SETTINGS
      mode = 'unordered';

CREATE TABLE stats (key UInt64, data String)
ENGINE = MergeTree() ORDER BY key;

CREATE MATERIALIZED VIEW consumer TO stats
AS SELECT key, data FROM azure_queue_engine_table;

SELECT * FROM stats ORDER BY key;

虚拟列

  • _path — 文件路径。
  • _file — 文件名。

有关虚拟列的更多信息，见此处。

内部信息

通过表设置 enable_logging_to_queue_log=1 为该表启用日志记录。

内部信息相关功能与 S3Queue 表引擎 相同，但有以下几个明显区别：

  1. 对于服务器版本 >= 25.1，使用 system.azure_queue_metadata_cache 作为队列的内存状态。对于更早的版本，使用 system.s3queue_metadata_cache (其中也会包含 azure 表的信息) 。
  2. 使用 system.azure_queue_metadata 表直接检查 Keeper 中存储的状态：每个元数据对象中 processed、processing 和 failed 节点的数量，以及按需查看其内容。这是 AzureQueue 对应的 system.s3_queue_metadata。
  3. 通过 ClickHouse 主配置启用 system.azure_queue_log，例如：
<azure_queue_log>
    <database>system</database>
    <table>azure_queue_log</table>
</azure_queue_log>

这个持久化表包含与 system.s3queue_metadata_cache 相同的信息，但记录的是已处理和处理失败的文件。

该表的结构如下：


CREATE TABLE system.azure_queue_log
(
    `hostname` LowCardinality(String) COMMENT 'Hostname',
    `event_date` Date COMMENT 'Event date of writing this log row',
    `event_time` DateTime COMMENT 'Event time of writing this log row',
    `database` String COMMENT 'The name of a database where current S3Queue table lives.',
    `table` String COMMENT 'The name of S3Queue table.',
    `uuid` String COMMENT 'The UUID of S3Queue table',
    `file_name` String COMMENT 'File name of the processing file',
    `rows_processed` UInt64 COMMENT 'Number of processed rows',
    `status` Enum8('Processed' = 0, 'Failed' = 1) COMMENT 'Status of the processing file',
    `processing_start_time` Nullable(DateTime) COMMENT 'Time of the start of processing the file',
    `processing_end_time` Nullable(DateTime) COMMENT 'Time of the end of processing the file',
    `exception` String COMMENT 'Exception message if happened'
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(event_date)
ORDER BY (event_date, event_time)
COMMENT 'Contains logging entries with the information files processes by S3Queue engine.'

示例：

SELECT *
FROM system.azure_queue_log
LIMIT 1
FORMAT Vertical

Row 1:
─

限制

AzureQueue 与 S3Queue 使用相同的实现，因此具有相同的限制。尤其是，ClickHouse 节点发生设备级断电时，已消费的行可能会在没有任何提示的情况下丢失：插入完成后，文件会立即在 Keeper 中标记为已处理 (若 after_processing = 'delete'，其源 blob 也会被删除) ，但插入的行只有在目标分片完成 fsync 后才会持久化；而默认情况下，fsync 不会同步执行 (fsync_after_insert = 0) 。对于推荐的物化视图消费方式，在目标 MergeTree 表上设置 fsync_after_insert = 1 (以及 fsync_part_directory = 1) 可大幅缩小这一时间窗口。

Navigation