CDC PostgreSQL с автоматическим созданием таблиц
Непрерывная загрузка Doris является экспериментальной функцией и поддерживается начиная с Doris 4.1.5.
Синхронизация с автоматическим созданием таблиц реализуется через собственный DDL FROM POSTGRES (...) TO DATABASE (...). Целевым объектом является база данных Doris, и Doris автоматически создает соответствующие целевые таблицы на основе схем исходных таблиц. С помощью include_tables можно указать, синхронизировать ли одну, несколько или все таблицы. При первой синхронизации Doris автоматически создает целевые таблицы с первичным ключом и сохраняет их первичные ключи согласованными с исходными. Такой подход подходит для сценариев зеркальной репликации, где обработка данных посредством SQL не требуется, а схемы целевых таблиц должны автоматически следовать схемам исходных таблиц.
Интегрируя возможности Flink CDC, Doris читает журналы изменений PostgreSQL и непрерывно записывает полные + инкрементальные данные группы таблиц в Doris через Stream Load. Синхронизация с автоматическим созданием таблиц позволяет назначить отдельной исходной таблице имя целевой таблицы или исключить столбцы, не являющиеся ключевыми. Если требуются SQL-выражения, фильтрация строк или преобразование данных, см. PostgreSQL CDC с сопоставлением SQL.
Сценарии использования
- Зеркальная репликация всей базы данных PostgreSQL или нескольких таблиц в Doris
- Поддержание схемы целевых таблиц согласованной с исходной без ручного создания таблиц
- Требуется только простое переименование целевых таблиц или удаление столбцов, без SQL-выражений, фильтрации строк или преобразования данных
- Нужны как полная инициализация, так и непрерывный захват инкрементальных изменений
Примечания
- В настоящее время гарантируется только семантика at-least-once.
- Синхронизировать можно только исходные таблицы с первичными ключами; автоматически создаваемые целевые таблицы используют модель Unique Key.
- Требуется привилегия Load. Если целевая таблица не существует, также требуется привилегия Create.
- При автоматическом создании таблиц уже существующая целевая таблица пропускается, поэтому пользователи могут настраивать таблицы под разные сценарии.
Быстрый запуск
Выполните приведенные ниже шаги, чтобы создать задание синхронизации PostgreSQL с автоматическим созданием таблиц и проверить его состояние.
Шаг 1. Создание задания загрузки
Используйте CREATE STREAMING JOB, чтобы создать задание непрерывной загрузки:
CREATE JOB test_postgres_job
ON STREAMING
FROM POSTGRES (
"jdbc_url" = "jdbc:postgresql://127.0.0.1:5432/postgres",
"driver_url" = "postgresql-42.5.1.jar",
"driver_class" = "org.postgresql.Driver",
"user" = "postgres",
"password" = "postgres",
"database" = "postgres",
"schema" = "public",
"include_tables" = "test_tbls",
"offset" = "latest"
)
TO DATABASE target_test_db (
"table.create.properties.replication_num" = "1" -- Set to 1 for single-BE deployments
)
Шаг 2. Проверка состояния загрузки
select * from jobs("type"="insert") where ExecuteType = "STREAMING";
Дополнительные общие операции (приостановка, возобновление, удаление, просмотр задач и т. д.) см. в разделе Обзор непрерывной загрузки.
Справочник параметров
Параметры источника (сторона PostgreSQL)
Параметры источника PostgreSQL настраивают подключение JDBC, область синхронизации и поведение разбиения полного снимка. Информация о подключении, сведения о драйвере, database и schema обязательны.
| Параметр | Обязательный | По умолчанию | Описание |
|---|---|---|---|
jdbc_url | Да | - | Строка подключения PostgreSQL JDBC. |
driver_url | Да | - | Путь к JAR-файлу драйвера JDBC. Поддерживаются имя файла, локальный абсолютный путь или URL HTTP. Подробнее см. в разделе Обзор JDBC Catalog. |
driver_class | Да | - | Имя класса драйвера JDBC, например org.postgresql.Driver. |
user | Да | - | Имя пользователя базы данных. |
password | Да | - | Пароль базы данных. |
database | Да | - | Имя базы данных PostgreSQL. Оно должно совпадать с именем базы данных в JDBC URL и после кодирования UTF-8 не может превышать 63 байта. |
schema | Да | - | Имя схемы. |
include_tables | Нет | - | Имена синхронизируемых таблиц через запятую. Если параметр не задан, синхронизируются все таблицы в схеме. |
exclude_tables | Нет | - | Имена таблиц, которые не нужно синхронизировать, через запятую. Параметр действует только если include_tables не задан. |
table.<table_name>.target_table | Нет | Имя исходной таблицы | Задает имя целевой таблицы Doris для исходной таблицы. <table_name> — имя исходной таблицы. |
table.<table_name>.exclude_columns | Нет | - | Задает исходные столбцы, которые не нужно синхронизировать, через запятую. Столбцы должны существовать и не могут включать столбцы первичного ключа. |
offset | Нет | latest | Начальная позиция. initial: полная и инкрементальная синхронизация; latest: синхронизация только изменений после запуска задания. Можно также указать точную позицию JSON, например {"lsn":"12345678"}. Поддерживается также snapshot, только для полной синхронизации. PostgreSQL не поддерживает earliest. |
snapshot_split_size | Нет | 40960 | Размер части в строках. При полной синхронизации таблица делится на несколько частей. Должно быть положительным целым числом. |
snapshot_parallelism | Нет | 1 | Параллелизм этапа полной синхронизации, то есть максимальное число частей, планируемых одной Task одновременно. Должно быть положительным целым числом. |
slot_name | Нет | doris_cdc_<job_id> | Имя слота логической репликации. Имя может содержать только строчные буквы, цифры и символы подчеркивания, не может начинаться с цифры и ограничено 63 символами. Пользовательский слот необходимо создать заранее; Doris его не удаляет. |
publication_name | Нет | doris_pub_<job_id> | Имя публикации. Правила именования такие же, как для slot_name. Пользовательскую публикацию необходимо создать заранее, включить в нее все синхронизируемые таблицы; Doris ее не удаляет. |
ssl_mode | Нет | disable | Режим SSL. Допустимые значения: disable, require и verify-ca. |
ssl_rootcert | Условно обязательный | - | Файл сертификата CA в формате FILE:<file_name>. Требуется, если ssl_mode имеет значение verify-ca. Сначала загрузите файл с помощью CREATE FILE. |
Если заданы и include_tables, и exclude_tables, приоритет имеет include_tables. Следующий фрагмент параметров FROM POSTGRES синхронизирует orders и customers, записывает orders в ods_orders, исключает неключевой столбец internal_note и использует заранее созданные ресурсы логической репликации с проверкой CA:
"include_tables" = "orders,customers",
"table.orders.target_table" = "ods_orders",
"table.orders.exclude_columns" = "internal_note",
"slot_name" = "orders_slot",
"publication_name" = "orders_publication",
"ssl_mode" = "verify-ca",
"ssl_rootcert" = "FILE:ca.pem"
Пользовательские слоты репликации и публикации управляются пользователем и сохраняются при удалении Streaming Job. Если slot_name или publication_name не указан, Doris создает соответствующий ресурс для задания и удаляет созданные им ресурсы при удалении задания.
Параметры целевой базы данных Doris
Параметры на стороне назначения управляют свойствами автоматически создаваемых таблиц и стратегией записи Stream Load. Все следующие параметры необязательны.
| Параметр | По умолчанию | Описание |
|---|---|---|
table.create.properties.* | - | Добавляет свойства таблицы при создании таблицы Doris, например table.create.properties.replication_num. |
load.strict_mode | false | Включать ли строгий режим для записей Stream Load. Допустимые значения: true и false. |
load.max_filter_ratio | 0 | Максимально допустимая доля фильтрации в окне выборки, в диапазоне [0, 1]. 0 означает, что строки с ошибками нельзя отфильтровать. Окно выборки составляет max_interval * 10 секунд. Задание приостанавливается, когда отношение числа ошибочных строк к общему числу строк в окне превышает это значение. |
Справочная информация
Синтаксис команды загрузки
Синтаксис создания задания синхронизации с автоматическим созданием таблиц выглядит следующим образом:
CREATE JOB <job_name>
[job_properties]
ON STREAMING
[ COMMENT <comment> ]
FROM POSTGRES (
[source_properties]
)
TO DATABASE <target_db> (
[target_properties]
)
Ниже описаны модули:
| Модуль | Описание |
|---|---|
| job_name | Имя задания |
| job_properties | Используется для задания общих параметров загрузки задания |
| comment | Используется для описания задания с помощью комментария |
| source_properties | Параметры, относящиеся к источнику PostgreSQL |
| target_properties | Параметры, относящиеся к целевой базе данных Doris |
job_properties поддерживает max_interval и compute_group. Подробнее см. в разделе Общие параметры конфигурации загрузки заданий. В режиме автоматического создания таблиц session.* не используется.
Часто задаваемые вопросы
В1: Как выбрать между синхронизацией с автоматическим созданием таблиц и синхронизацией с сопоставлением SQL?
- Если требуется зеркальная репликация, автоматическое создание таблиц или только переименование целевых таблиц и исключение неключевых столбцов, используйте синхронизацию с автоматическим созданием таблиц.
- Если в конвейере синхронизации нужны SQL-выражения, фильтрация строк или преобразование данных, используйте PostgreSQL CDC с сопоставлением SQL.
В2: Поддерживаются ли таблицы без первичного ключа?
В настоящее время поддерживается только синхронизация таблиц с первичным ключом. Таблицы без первичного ключа пока не поддерживаются.
В3: Что делать, если создание таблицы завершается ошибкой при развертывании с одним BE?
Явно задайте "table.create.properties.replication_num" = "1" в предложении TO DATABASE, чтобы избежать несоответствия между количеством реплик по умолчанию и числом доступных BE.
В4: Если целевая таблица уже существует, будет ли она перезаписана?
Нет. На этапе автоматического создания таблиц существующие целевые таблицы пропускаются. При необходимости можно заранее настроить схему таблицы.
В5: Как синхронизировать только инкрементальные данные и пропустить этап полной синхронизации?
Установите для offset значение latest. Задание будет получать только последние инкрементальные изменения и больше не будет выполнять полную инициализацию.