motor de tabla S3Queue - ClickHouse Documentation

Este motor proporciona integración con el ecosistema de Amazon S3 y permite la importación en streaming. Este motor es similar a los motores Kafka y RabbitMQ, pero ofrece características específicas de S3. Es importante entender esta nota del PR original de la implementación de S3Queue: cuando se vincula una MATERIALIZED VIEW al motor, el motor de tabla S3Queue comienza a recopilar datos en segundo plano.

CREATE table

CREATE TABLE s3_queue_engine_table (name String, value UInt32)
    ENGINE = S3Queue(path, [NOSIGN, | aws_access_key_id, aws_secret_access_key,] format, [compression], [headers], [extra_credentials])
    [SETTINGS]
    [mode = '',]
    [after_processing = 'keep',]
    [keeper_path = '',]
    [loading_retries = 0,]
    [processing_threads_num = 16,]
    [parallel_inserts = false,]
    [enable_logging_to_queue_log = true,]
    [last_processed_path = "",]
    [tracked_files_limit = 1000,]
    [tracked_file_ttl_sec = 0,]
    [polling_min_timeout_ms = 1000,]
    [polling_max_timeout_ms = 10000,]
    [polling_backoff_ms = 0,]
    [cleanup_interval_min_ms = 10000,]
    [cleanup_interval_max_ms = 30000,]
    [buckets = 0,]
    [list_objects_batch_size = 1000,]
    [enable_hash_ring_filtering = 0,]
    [max_processed_files_before_commit = 100,]
    [max_processed_rows_before_commit = 0,]
    [max_processed_bytes_before_commit = 0,]
    [max_processing_time_sec_before_commit = 0,]

Antes de la versión 24.7, es obligatorio usar el prefijo s3queue_ para todos los ajustes, excepto mode, after_processing y keeper_path.

Parámetros del motor Los parámetros de S3Queue son los mismos que admite el motor de tabla S3. Consulte la sección de parámetros aquí. Ejemplo

CREATE TABLE s3queue_engine_table (name String, value UInt32)
ENGINE=S3Queue('https://clickhouse-public-datasets.s3.amazonaws.com/my-test-bucket-768/*', 'CSV', 'gzip')
SETTINGS
    mode = 'unordered';

Uso de colecciones nombradas:

<clickhouse>
    <named_collections>
        <s3queue_conf>
            <url>'https://clickhouse-public-datasets.s3.amazonaws.com/my-test-bucket-768/*</url>
            <access_key_id>test<access_key_id>
            <secret_access_key>test</secret_access_key>
        </s3queue_conf>
    </named_collections>
</clickhouse>

CREATE TABLE s3queue_engine_table (name String, value UInt32)
ENGINE=S3Queue(s3queue_conf, format = 'CSV', compression_method = 'gzip')
SETTINGS
    mode = 'ordered';

Configuración

Para obtener una lista de la configuración de la tabla, use la tabla system.s3_queue_settings. Disponible a partir de la versión 24.10.

Nombres de configuración (24.7+)A partir de la versión 24.7, la configuración de S3Queue puede especificarse con o sin el prefijo s3queue_:

Sintaxis moderna (24.7+): processing_threads_num, tracked_file_ttl_sec, etc.
Sintaxis heredada (todas las versiones): s3queue_processing_threads_num, s3queue_tracked_file_ttl_sec, etc.

Ambas formas son compatibles a partir de la versión 24.7. Los ejemplos de esta página usan la sintaxis moderna, sin prefijo.

Modo

Valores posibles:

unordered — En el modo no ordenado, el conjunto de todos los archivos ya procesados se registra mediante nodos persistentes en ZooKeeper.
ordered — En el modo ordenado, los archivos se procesan en orden lexicográfico. Esto significa que, si en algún momento se procesó un archivo llamado ‘BBB’ y más tarde se agrega al bucket un archivo llamado ‘AA’, este se ignorará. En ZooKeeper solo se almacenan el nombre máximo (en sentido lexicográfico) del archivo consumido correctamente y los nombres de los archivos que se volverán a intentar tras un intento de carga fallido.

Valor predeterminado: ordered en las versiones anteriores a la 24.6. A partir de la 24.6, no hay valor predeterminado; la configuración pasa a ser obligatoria y debe especificarse manualmente. En las tablas creadas en versiones anteriores, el valor predeterminado seguirá siendo Ordered por compatibilidad.

`after_processing`

Cómo gestionar el archivo después de procesarlo correctamente. Valores posibles:

keep.
delete.
move.
tag.

Valor predeterminado: keep. move requiere configuración adicional. En caso de moverlo dentro del mismo bucket, se debe proporcionar un nuevo prefijo de ruta como after_processing_move_prefix. Para moverlo a otro bucket de S3, se requiere el URI del bucket de destino como after_processing_move_uri y las credenciales de S3 como after_processing_move_access_key_id y after_processing_move_secret_access_key. Ejemplo:

CREATE TABLE s3queue_engine_table (name String, value UInt32)
ENGINE=S3Queue('https://clickhouse-public-datasets.s3.amazonaws.com/my-test-bucket-768/*', 'CSV', 'gzip')
SETTINGS
    mode = 'unordered',
    after_processing = 'move',
    after_processing_retries = 20,
    after_processing_move_prefix = 'dst_prefix',
    after_processing_move_uri = 'https://clickhouse-public-datasets.s3.amazonaws.com/dst-bucket',
    after_processing_move_access_key_id = 'test',
    after_processing_move_secret_access_key = 'test';

Para mover datos de un contenedor de Azure a otro, se requiere la connection string de Blob Storage como after_processing_move_connection_string y el nombre del contenedor como after_processing_move_container. Consulte la configuración de AzureQueue. El etiquetado requiere la clave y el valor de la etiqueta, proporcionados como after_processing_tag_key y after_processing_tag_value.

`after_processing_retries`

Número de reintentos para la acción solicitada después del procesamiento, antes de desistir. Valores posibles:

Entero no negativo.

Valor predeterminado: 10.

`after_processing_move_access_key_id`

ID de la clave de acceso para el bucket de S3 al que se moverán los archivos procesados correctamente, si el destino es otro bucket de S3. Valores posibles:

String.