Moteur de table AzureQueue

Ce moteur s'intègre à l'écosystème Azure Blob Storage, permettant l'importation de données en streaming.

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

Paramètres du moteur

Les paramètres de AzureQueue sont les mêmes que ceux pris en charge par le moteur de table AzureBlobStorage. Consultez la section des paramètres ici.

Comme avec le moteur de table AzureBlobStorage, les utilisateurs peuvent utiliser l’émulateur Azurite pour le développement local avec Azure Storage. Pour plus de détails, cliquez ici.

Exemple

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'

L’ensemble des paramètres pris en charge est pour l’essentiel le même que pour le moteur de table S3Queue, mais sans le préfixe s3queue_. Voir la liste complète des paramètres. Pour obtenir la liste des paramètres configurés pour la table, utilisez la table system.azure_queue_settings. Disponible à partir de 24.10.

Vous trouverez ci-dessous les paramètres compatibles uniquement avec AzureQueue et non applicables à S3Queue.

Chaîne de connexion à Azure Blob Storage vers laquelle déplacer les fichiers traités avec succès, si la destination est un autre conteneur Azure.

Valeurs possibles :

• String.

Valeur par défaut : chaîne vide.

Nom du conteneur vers lequel déplacer les fichiers traités avec succès si la destination est un autre conteneur Azure.

Valeurs possibles :

• String.

Valeur par défaut : chaîne vide.

Exemple :

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

Les requêtes SELECT sont interdites par défaut sur les tables AzureQueue. Cela suit le modèle classique des files d'attente, où les données sont lues une seule fois, puis retirées de la file d'attente. SELECT est interdit afin d'éviter toute perte accidentelle de données. Cependant, cela peut parfois s'avérer utile. Pour ce faire, vous devez définir le paramètre stream_like_engine_allow_direct_select sur True. Le moteur AzureQueue dispose d'un paramètre spécial pour les requêtes SELECT : commit_on_select. Définissez-le sur False pour conserver les données dans la file d'attente après leur lecture, ou sur True pour les supprimer.

SELECT n’est pas particulièrement utile pour l’importation en continu (sauf pour le débogage), car chaque fichier ne peut être importé qu’une seule fois. Il est plus pratique de mettre en place un traitement en temps réel à l’aide de vues matérialisées. Pour ce faire :

1. Utilisez le moteur pour créer une table qui lit depuis le chemin spécifié dans Azure Blob Storage et traitez-la comme un flux de données.
2. Créez une table avec la structure souhaitée.
3. Créez une vue matérialisée qui convertit les données du moteur et les insère dans une table créée précédemment.

Lorsque la MATERIALIZED VIEW est associée au moteur, elle commence à collecter les données en arrière-plan.

Les arguments du moteur ont la forme AzureQueue(connection_string, container_name, blobpath, format[, compression]).

Exemple :

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 — Chemin d’accès au fichier.
• _file — Nom du fichier.

Pour plus d’informations sur les colonnes virtuelles, consultez ici.

Activez la journalisation pour la table via le paramètre de table enable_logging_to_queue_log=1.

Les capacités d'introspection sont les mêmes que pour le moteur de table S3Queue, avec toutefois quelques différences notables :

1. Utilisez system.azure_queue_metadata_cache pour l'état en mémoire de la file d'attente pour les versions du serveur >= 25.1. Pour les versions antérieures, utilisez system.s3queue_metadata_cache (il contiendrait également des informations sur les tables azure).
2. Utilisez la table system.azure_queue_metadata pour inspecter directement l'état stocké dans Keeper : le nombre de nœuds processed, processing et failed par objet de métadonnées et, à la demande, leur contenu. Il s'agit de l'équivalent AzureQueue de system.s3_queue_metadata.
3. Activez system.azure_queue_log via la configuration principale de ClickHouse, par exemple.

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

Cette table persistante contient les mêmes informations que system.s3queue_metadata_cache, mais pour les fichiers traités et ceux ayant échoué.

La table a la structure suivante :

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

Exemple :

SELECT *
FROM system.azure_queue_log
LIMIT 1
FORMAT Vertical

Row 1:
─

AzureQueue partage la même implémentation que S3Queue et présente les mêmes limitations. En particulier, une coupure d'alimentation au niveau matériel du nœud ClickHouse peut entraîner la perte silencieuse de lignes consommées : un fichier est enregistré comme traité dans Keeper (et, avec after_processing = 'delete', son blob source est supprimé) dès la fin de l’insert, mais les lignes insérées ne sont durablement écrites qu’une fois la part cible synchronisée avec fsync, ce qui ne se produit pas de manière synchrone par défaut (fsync_after_insert = 0). Pour le chemin de consommation recommandé via une vue matérialisée, définir fsync_after_insert = 1 (et fsync_part_directory = 1) sur la table MergeTree cible réduit considérablement cette fenêtre.