Motor de tabla AzureQueue

Este motor ofrece integración con el ecosistema de Azure Blob Storage, lo que permite importar datos en streaming.

CREATE TABLE

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

Parámetros del motor Los parámetros de AzureQueue son los mismos que admite el motor de tabla AzureBlobStorage. Consulte la sección de parámetros aquí. Al igual que con el motor de tabla AzureBlobStorage, los usuarios pueden usar el emulador Azurite para el desarrollo local con Azure Storage. Encontrará más información aquí. Ejemplo

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'

Configuración

El conjunto de ajustes admitidos es prácticamente el mismo que para el motor de tabla S3Queue, pero sin el prefijo s3queue_. Consulte la lista completa de ajustes. Para obtener una lista de los ajustes configurados para la tabla, use la tabla system.azure_queue_settings. Disponible a partir de la versión 24.10. A continuación se muestran los ajustes compatibles únicamente con AzureQueue y no aplicables a S3Queue.

`after_processing_move_connection_string`

Cadena de conexión de Azure Blob Storage a la que mover los archivos procesados correctamente, si el destino es otro contenedor de Azure. Valores posibles:

String.

Valor predeterminado: cadena vacía.

`after_processing_move_container`

Nombre del contenedor al que se moverán los archivos procesados correctamente si el destino es otro contenedor de Azure. Valores posibles:

String.

Valor predeterminado: cadena vacía. Ejemplo:

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';

SELECT en el motor de tabla AzureQueue

Las consultas SELECT están prohibidas de forma predeterminada en las tablas AzureQueue. Esto sigue el patrón común de las colas, en el que los datos se leen una vez y luego se eliminan de la cola. SELECT está prohibido para evitar la pérdida accidental de datos. Sin embargo, en algunos casos puede resultar útil. Para ello, debe establecer la SETTING stream_like_engine_allow_direct_select en True. El motor AzureQueue tiene una SETTING especial para las consultas SELECT: commit_on_select. Establézcala en False para conservar los datos en la cola después de leerlos, o en True para eliminarlos. (Nota: esta SETTING no tiene sentido en el modo exclusive y se ignora; el modo exclusive siempre actúa como si commit_on_select fuera True.)

Descripción

SELECT no resulta especialmente útil para la importación en streaming (salvo para tareas de depuración), porque cada archivo solo puede importarse una vez. Es más práctico crear flujos en tiempo real mediante vistas materializadas. Para ello:

Use el motor para crear una tabla que consuma desde la ruta especificada en Azure Blob Storage y trátela como un flujo de datos.
Cree una tabla con la estructura deseada.
Cree una vista materializada que convierta los datos del motor y los inserte en una tabla creada previamente.

Cuando la MATERIALIZED VIEW se conecta al motor, empieza a recopilar datos en segundo plano. Los argumentos del motor tienen la forma AzureQueue(connection_string, container_name, blobpath, format[, compression]). Ejemplo:

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;

Columnas virtuales

_path — Ruta del archivo.
_file — Nombre del archivo.

Para obtener más información sobre las columnas virtuales, consulte aquí.

Introspección

Habilite el registro de la tabla mediante la configuración de tabla enable_logging_to_queue_log=1. Las capacidades de introspección son las mismas que las del motor de tabla S3Queue, con varias diferencias concretas:

Use system.azure_queue_metadata_cache para el estado en memoria de la cola en las versiones del servidor >= 25.1. Para versiones anteriores, use system.s3queue_metadata_cache (también contendría información de las tablas azure).
Use la tabla system.azure_queue_metadata para inspeccionar directamente el estado almacenado en Keeper: el número de nodos processed, processing y failed por objeto de metadatos y, cuando se solicite, su contenido. Esta es la contraparte de AzureQueue de system.s3_queue_metadata.
Habilite system.azure_queue_log mediante la configuración principal de ClickHouse; por ejemplo:

<azure_queue_log>
    <database>system</database>
    <table>azure_queue_log</table>
</azure_queue_log>

Esta tabla persistente contiene la misma información que system.s3queue_metadata_cache, pero sobre los archivos procesados y fallidos. La tabla tiene la siguiente estructura:

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.'

Ejemplo:

SELECT *
FROM system.azure_queue_log
LIMIT 1
FORMAT Vertical

Row 1:
──────
hostname:              clickhouse
event_date:            2024-12-16
event_time:            2024-12-16 13:42:47
database:              default
table:                 azure_queue_engine_table
uuid:                  1bc52858-00c0-420d-8d03-ac3f189f27c8
file_name:             test_1.csv
rows_processed:        3
status:                Processed
processing_start_time: 2024-12-16 13:42:47
processing_end_time:   2024-12-16 13:42:47
exception:

1 row in set. Elapsed: 0.002 sec.

Limitaciones

AzureQueue comparte la misma implementación que S3Queue y presenta las mismas limitaciones. En particular, una pérdida de alimentación del dispositivo en el que se ejecuta el nodo de ClickHouse puede causar la pérdida silenciosa de filas consumidas: un archivo se registra como procesado en Keeper (y, con after_processing = 'delete', se elimina su blob de origen) en cuanto finaliza la inserción, pero las filas insertadas solo son duraderas una vez que se sincroniza mediante fsync la parte de destino, lo que no ocurre de forma síncrona de manera predeterminada (fsync_after_insert = 0). En la ruta de consumo recomendada mediante vista materializada, configurar fsync_after_insert = 1 (y fsync_part_directory = 1) en la tabla MergeTree de destino reduce considerablemente este intervalo.

Referencia de SQL

Tipos de datos

Motores

Funciones

Formatos

Configuración

Tablas del sistema

Lagos de datos

CREATE TABLE

Configuración

`after_processing_move_connection_string`

`after_processing_move_container`

SELECT en el motor de tabla AzureQueue

Descripción

Columnas virtuales

Introspección

Limitaciones

​CREATE TABLE

​Configuración

​after_processing_move_connection_string

​after_processing_move_container

​SELECT en el motor de tabla AzureQueue

​Descripción

​Columnas virtuales

​Introspección

​Limitaciones

CREATE TABLE

Configuración

`after_processing_move_connection_string`

`after_processing_move_container`

SELECT en el motor de tabla AzureQueue

Descripción

Columnas virtuales

Introspección

Limitaciones