движок таблицы S3Queue - ClickHouse Documentation

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

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

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

Параметры движка Параметры 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';

Настройки

Чтобы получить список настроек, заданных для таблицы, используйте таблицу 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 и выше. В примерах на этой странице используется современный синтаксис без префикса.

Режим

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

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`

Идентификатор ключа доступа для S3 бакета, в который перемещаются успешно обработанные файлы, если пункт назначения — другой S3 бакет. Возможные значения:

String.