Движок таблицы S3Queue

Этот движок обеспечивает интеграцию с экосистемой Amazon S3 и позволяет выполнять потоковый импорт. Он аналогичен движкам Kafka и RabbitMQ, но предоставляет функции, специфичные для S3.

Важно учитывать следующее примечание из исходного PR по реализации S3Queue: когда к таблице с этим движком присоединяется MATERIALIZED VIEW, движок таблицы S3Queue начинает собирать данные в фоновом режиме.

Создать таблицу

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])
    [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,]

Примечание

До версии 24.7 требуется использовать префикс s3queue_ для всех настроек, кроме mode, after_processing и keeper_path.

Параметры движка

Параметры S3Queue такие же, как у табличного движка S3. См. раздел «Параметры» здесь.

Пример

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

Использование именованных коллекций:

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

Settings

Чтобы получить список настроек, заданных для таблицы, используйте таблицу system.s3_queue_settings. Доступно, начиная с версии 24.10.

Имена настроек (24.7+)

Начиная с версии 24.7, настройки S3Queue можно указывать с префиксом s3queue_ или без него:

Современный синтаксис (24.7+): processing_threads_num, tracked_file_ttl_sec и т. д.
Устаревший синтаксис (все версии): s3queue_processing_threads_num, s3queue_tracked_file_ttl_sec и т. д.

Оба варианта поддерживаются в 24.7+. Примеры на этой странице используют современный синтаксис без префикса.

Mode

Возможные значения:

unordered — В режиме unordered множество всех уже обработанных файлов отслеживается с помощью постоянных узлов в ZooKeeper.
ordered — В режиме ordered файлы обрабатываются в лексикографическом порядке. Это означает, что если файл с именем BBB был обработан в какой‑то момент, а позже в бакет был добавлен файл с именем AA, он будет проигнорирован. В ZooKeeper сохраняются только максимальное имя (в лексикографическом смысле) успешно обработанного файла и имена файлов, которые будут повторно загружены после неудачной попытки загрузки.

Значение по умолчанию: ordered в версиях до 24.6. Начиная с 24.6 значение по умолчанию отсутствует, настройку требуется указывать вручную. Для таблиц, созданных в более ранних версиях, значение по умолчанию останется ordered для сохранения совместимости.

`after_processing`

Что делать с файлом после успешной обработки.

Возможные значения:

keep.
delete.
move.
tag.

Значение по умолчанию: keep.

Для варианта move требуются дополнительные настройки. В случае перемещения в пределах того же бакета необходимо указать новый префикс пути в параметре after_processing_move_prefix.

Перемещение в другой S3‑бакет требует указания URI целевого бакета в параметре after_processing_move_uri, а также учетных данных доступа к S3 в параметрах after_processing_move_access_key_id и after_processing_move_secret_access_key.

Пример:

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

Для перемещения данных из одного контейнера Azure в другой необходимо указать строку подключения Blob Storage в параметре after_processing_move_connection_string и имя контейнера в параметре after_processing_move_container. См. настройки AzureQueue.

Для добавления тегов необходимо указать ключ и значение тега в параметрах after_processing_tag_key и after_processing_tag_value.

`after_processing_retries`

Количество повторных попыток выполнения запрошенного действия послеобработки, после которых попытки прекращаются.

Возможные значения:

Неотрицательное целое число.

Значение по умолчанию: 10.

`after_processing_move_access_key_id`

ID ключа доступа (Access Key ID) для S3‑бакета, в который нужно переместить успешно обработанные файлы, если целевым местом назначения является другой S3‑бакет.

Возможные значения:

Строка.