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

Этот движок обеспечивает интеграцию в режиме только для чтения с существующими таблицами Apache Paimon, хранящимися в Amazon S3, Azure, HDFS и локально. Он поддерживает чтение снимка, инкрементальное чтение и базовое отсечение партиций, реализованное в движке.

Обратите внимание, что таблица Paimon уже должна существовать в хранилище; эта команда не поддерживает DDL-параметры для создания новой таблицы. Создание таблиц Paimon* регулируется параметром allow_experimental_paimon_storage_engine (по умолчанию он отключен), поэтому включите его перед выполнением CREATE TABLE.

SET allow_experimental_paimon_storage_engine = 1;

CREATE TABLE paimon_table_s3
    ENGINE = PaimonS3(url, [, access_key_id, secret_access_key] [,format] [,compression])

CREATE TABLE paimon_table_azure
    ENGINE = PaimonAzure(connection_string|storage_account_url, container_name, blobpath, [,account_name], [,account_key] [,format] [,compression_method])

CREATE TABLE paimon_table_hdfs
    ENGINE = PaimonHDFS(path_to_table, [,format] [,compression_method])

CREATE TABLE paimon_table_local
    ENGINE = PaimonLocal(path_to_table, [,format] [,compression_method])

Описание аргументов совпадает с описанием аргументов для движков S3, AzureBlobStorage, HDFS и File соответственно. format обозначает формат файлов данных в таблице Paimon.

Параметры движка можно задавать с помощью именованных коллекций

CREATE TABLE paimon_table ENGINE=PaimonS3('http://test.s3.amazonaws.com/clickhouse-bucket/test_table', 'test', 'test')

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

<clickhouse>
    <named_collections>
        <paimon_conf>
            <url>http://test.s3.amazonaws.com/clickhouse-bucket/</url>
            <access_key_id>test</access_key_id>
            <secret_access_key>test</secret_access_key>
        </paimon_conf>
    </named_collections>
</clickhouse>

CREATE TABLE paimon_table ENGINE=PaimonS3(paimon_conf, filename = 'test_table')

• Чтение снимка из последнего снимка таблицы.
• Инкрементальное чтение по идентификатору зафиксированного снимка, если эта возможность включена.
• Отсечение партиций, если включен use_paimon_partition_pruning.
• Необязательное фоновое обновление метаданных при соответствующей настройке.
• Стабильный UUID таблицы при использовании баз данных Atomic/Replicated, что позволяет использовать макросы {uuid} в путях Keeper.

Этот движок использает те же настройки, что и соответствующие движки объектного хранилища, а также добавляет настройки, специфичные для Paimon:

• allow_experimental_paimon_storage_engine — включает создание движков таблиц Paimon, PaimonS3, PaimonAzure, PaimonHDFS и PaimonLocal. По умолчанию: 0 (отключено).
• use_paimon_metadata_files_cache — включает кэш файлов метаданных Paimon (кэширует десериализованные списки manifest и manifest-файлы). Установите 1, чтобы включить, 0 — чтобы отключить. По умолчанию: 0. Применение этой настройки различается для табличных функций и движков постоянных таблиц — см. примечание ниже.
• paimon_incremental_read — включает режим инкрементального чтения.
• paimon_metadata_refresh_interval_sec — интервал фонового обновления метаданных в секундах. Если задано значение больше 0, фоновая задача периодически загружает из объектного хранилища последний снимок и схему. По умолчанию: 30.
• paimon_keeper_path — путь Keeper для состояния инкрементального чтения. Должен быть задан и быть уникальным для каждой таблицы; поддерживает макросы, такие как {database}, {table}, {uuid}.
• paimon_replica_name — имя реплики для состояния инкрементального чтения. Должно быть задано и быть уникальным для каждой реплики; поддерживает макросы, такие как {replica}.

Пример (включение экспериментального движка Paimon и кэша файлов метаданных):

SET allow_experimental_paimon_storage_engine = 1;
SET use_paimon_metadata_files_cache = 1;

CREATE TABLE paimon_cached
ENGINE = PaimonS3(paimon_conf, filename = 'paimon_all_types');

Инкрементальное чтение с состоянием в Keeper:

CREATE TABLE paimon_inc
ENGINE = PaimonS3(paimon_conf, filename = 'paimon_all_types')
SETTINGS
    paimon_incremental_read = 1,
    paimon_keeper_path = '/clickhouse/{database}/{uuid}',
    paimon_replica_name = '{replica}';

Следующие настройки действуют на уровне запроса (передаются через SELECT ... SETTINGS, а не в CREATE TABLE). Они управляют поведением инкрементального чтения в рамках отдельного запроса:

• paimon_target_snapshot_id — считывать только дельту указанного снимка. Зафиксированная водяная метка в Keeper не продвигается, поэтому один и тот же снимок можно перечитывать сколько угодно раз. По умолчанию: -1 (отключено).
• max_consume_snapshots — максимальное число снимков, считываемых за одно инкрементальное чтение. Если в источнике накопилось много непрочитанных снимков, этот параметр ограничивает, сколько из них будет считано за один запрос, чтобы контролировать размер батча. 0 означает отсутствие ограничений. По умолчанию: 0.

Чтение указанного снимка — всегда возвращает дельту снимка 1 независимо от текущей водяной метки:

SELECT count()
FROM paimon_inc
SETTINGS paimon_target_snapshot_id = 1;

Ограничение числа снимков в батче — если ожидают обработки три новых снимка, обрабатывайте не более двух за один запрос:

SELECT count()
FROM paimon_inc
SETTINGS max_consume_snapshots = 2;

Вы можете построить сквозной конвейер, который непрерывно синхронизирует данные из таблицы Paimon в таблицу семейства MergeTree с помощью refreshable materialized view в режиме APPEND. В каждом цикле обновления считываются только новые инкрементальные данные из Paimon и добавляются в целевую таблицу.

Шаг 1 — Создайте исходную таблицу Paimon с включенными инкрементальным чтением и обновлением метаданных.

В примере ниже используется PaimonLocal. Замените движок на PaimonS3, PaimonAzure, PaimonHDFS или движок Paimon с автоопределением в зависимости от используемого backend-хранилища:

SET allow_experimental_paimon_storage_engine = 1;

-- Local storage
CREATE TABLE paimon_mv_source
ENGINE = PaimonLocal('/path/to/paimon/table')
SETTINGS
    paimon_incremental_read = 1,
    paimon_keeper_path = '/clickhouse/tables/{uuid}',
    paimon_replica_name = '{replica}',
    paimon_metadata_refresh_interval_sec = 1;

-- S3 storage (the `Paimon` engine defaults to the S3 implementation when no `disk` is specified)
CREATE TABLE paimon_mv_source
ENGINE = Paimon('http://minio:9000/bucket/path/to/table', 'access_key', 'secret_key')
SETTINGS
    paimon_incremental_read = 1,
    paimon_keeper_path = '/clickhouse/tables/{uuid}',
    paimon_replica_name = '{replica}',
    paimon_metadata_refresh_interval_sec = 1;

paimon_metadata_refresh_interval_sec задает интервал фонового обновления метаданных в секундах. Если значение больше 0, фоновая задача периодически подтягивает актуальные снимок и схему из Объектного хранилища, чтобы цикл обновления MV мог видеть недавно зафиксированные данные без ожидания запроса, который запустит обновление метаданных. Значение по умолчанию — 30. Используйте с осторожностью при работе с большим числом таблиц, чтобы избежать чрезмерной нагрузки на Объектное хранилище и операции ввода-вывода Keeper.

Шаг 2 — Создайте целевую таблицу MergeTree (схема скопирована из таблицы Paimon):

CREATE TABLE paimon_mv_dest AS paimon_mv_source
ENGINE = MergeTree()
ORDER BY tuple();

Шаг 3 — Создайте refreshable Materialized View:

CREATE MATERIALIZED VIEW paimon_mv
REFRESH EVERY 10 SECOND
APPEND
TO paimon_mv_dest
AS SELECT * FROM paimon_mv_source;

Каждые 10 секунд MV выполняет SELECT * FROM paimon_mv_source, который возвращает только строки, добавленные после последнего зафиксированного снимка, и дописывает их в paimon_mv_dest.

Очистка:

SYSTEM STOP VIEW paimon_mv;
DROP VIEW IF EXISTS paimon_mv SYNC;
DROP TABLE IF EXISTS paimon_mv_dest SYNC;
DROP TABLE IF EXISTS paimon_mv_source SYNC;

• Для инкрементального чтения должен быть настроен Keeper (ZooKeeper).
• Для инкрементального чтения параметр paimon_keeper_path должен быть задан и быть уникальным для каждой таблицы.
• paimon_replica_name должен быть уникальным для каждой реплики в рамках одного и того же пути Keeper.
• Инкрементальное чтение использует доставку не более одного раза: зафиксированный снимок продвигается в момент сбора файлов данных, ещё до фактического чтения данных. Если запрос завершится ошибкой после сбора файлов, пропущенные снимки не будут повторно прочитаны при повторной попытке.
• Движок таблицы работает в режиме только для чтения; изменение данных не поддерживается.
• Инкрементальное чтение не обрабатывает удаление исторических данных из источника Paimon. Если исходные данные Paimon удаляются или обновляются, соответствующие строки, уже записанные в целевую таблицу семейства MergeTree в ClickHouse, не будут автоматически удалены. Чтобы очистить устаревшие данные, необходимо вручную выполнить ALTER TABLE ... DELETE для таблицы семейства MergeTree.
• Если базовая таблица Paimon удаляется и создаётся заново по тому же пути объектного хранилища (например, через Flink или Spark), необходимо выполнить DROP и повторно CREATE соответствующую таблицу ClickHouse. ClickHouse обнаруживает повторное создание, сравнивая временную метку создания schema-0, и выдаёт ошибку; устаревшую таблицу ClickHouse нельзя использовать, пока она не будет создана заново.

Движок таблицы Paimon автоматически определяет тип хранилища по настройке disk и в зависимости от этого использует PaimonS3, PaimonAzure или PaimonLocal. Если disk не указан, по умолчанию используется реализация PaimonS3.

• _path — Путь к файлу. Тип: LowCardinality(String).
• _file — Имя файла. Тип: LowCardinality(String).
• _size — Размер файла в байтах. Тип: Nullable(UInt64). Если размер файла неизвестен, значение — NULL.
• _time — Время последнего изменения файла. Тип: Nullable(DateTime). Если время неизвестно, значение — NULL.
• _etag — ETag файла. Тип: LowCardinality(String). Если ETag неизвестен, значение — NULL.

Типы данных, поддерживаемые в ключах партиционирования Paimon:

• CHAR
• VARCHAR
• BOOLEAN
• DECIMAL
• TINYINT
• SMALLINT
• INTEGER
• DATE
• TIME
• TIMESTAMP
• TIMESTAMP WITH LOCAL TIME ZONE
• BIGINT
• FLOAT
• DOUBLE