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. Если во время синхронизации необходимо выполнять сопоставление столбцов, фильтрацию или преобразование данных, см. PostgreSQL CDC с SQL-сопоставлением.
Сценарии использования
- Зеркальная репликация всей базы данных PostgreSQL или нескольких таблиц в Doris
- Сохранение согласованности схемы целевых таблиц с источником без ручного создания таблиц
- Отсутствие необходимости выполнять отбор столбцов, фильтрацию или преобразование данных в конвейере синхронизации
- Необходимость как полной инициализации, так и непрерывного захвата инкрементальных изменений
Примечания
- В настоящее время гарантируется только семантика at-least-once.
- В настоящее время поддерживается синхронизация только таблиц с первичным ключом.
- Требуется привилегия 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-соединение, область синхронизации и поведение разбиения полного снимка.
| Параметр | По умолчанию | Описание |
|---|---|---|
| jdbc_url | - | Строка подключения PostgreSQL JDBC |
| driver_url | - | Путь к JAR-файлу драйвера JDBC. Поддерживаются имя файла, локальный абсолютный путь и URL HTTP. Подробнее см. Обзор JDBC Catalog |
| driver_class | - | Имя класса драйвера JDBC |
| user | - | Имя пользователя базы данных |
| password | - | Пароль базы данных |
| database | - | Имя базы данных |
| schema | - | Имя схемы |
| include_tables | - | Таблицы для синхронизации, разделённые запятыми. Если оставить пустым, по умолчанию синхронизируются все таблицы |
| offset | latest | latest: только инкрементальная синхронизация; initial: полная + инкрементальная синхронизация |
| snapshot_split_size | 40960 | Размер одного разбиения (в строках). Во время полной синхронизации таблица делится на несколько разбиений для синхронизации |
| snapshot_parallelism | 1 | Параллелизм на этапе полной синхронизации, то есть максимальное число разбиений, запланированных в одном Task |
Параметры целевой базы данных Doris
Параметры целевой стороны управляют свойствами автоматически создаваемых таблиц и стратегией записи Stream Load.
| Параметр | По умолчанию | Описание |
|---|---|---|
| table.create.properties.* | - | Поддерживает указание свойств таблицы при её создании, например replication_num |
| load.strict_mode | - | Включать ли строгий режим. По умолчанию отключён |
| load.max_filter_ratio | - | Максимально допустимый коэффициент фильтрации в пределах окна выборки. Значение должно находиться в диапазоне от 0 до 1 включительно. Значение по умолчанию — 0, то есть фильтрация не допускается. Окно выборки равно max_interval * 10. Если в пределах окна выборки отношение числа ошибочных строк к общему числу строк превышает max_filter_ratio, регулярное задание приостанавливается; для проверки проблем с качеством данных требуется вмешательство пользователя. |
Справочник
Синтаксис команды загрузки
Синтаксис создания задания синхронизации с автоматическим созданием таблиц имеет следующий вид:
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 |
Часто задаваемые вопросы
Q1: Как выбрать между синхронизацией с автоматическим созданием таблиц и синхронизацией с SQL-сопоставлением?
- Когда требуется зеркальная репликация, автоматическое создание таблиц и не требуется отбор или преобразование столбцов, используйте синхронизацию с автоматическим созданием таблиц.
- Когда нужно выполнять сопоставление столбцов, фильтрацию или преобразование данных в конвейере синхронизации, используйте PostgreSQL CDC с SQL-сопоставлением.
Q2: Поддерживаются ли таблицы без первичного ключа?
В настоящее время поддерживается только синхронизация таблиц с первичным ключом. Таблицы без первичного ключа пока не поддерживаются.
Q3: Что делать, если создание таблицы завершается ошибкой в развертывании с одним BE?
Необходимо явно задать "table.create.properties.replication_num" = "1" в предложении TO DATABASE, чтобы избежать несоответствия между числом реплик по умолчанию и количеством доступных BE.
Q4: Если целевая таблица уже существует, будет ли она перезаписана?
Нет. На этапе автоматического создания таблиц существующие целевые таблицы пропускаются. При необходимости можно заранее настроить схему таблицы.
Q5: Как синхронизировать только инкрементальные данные и пропустить этап полной синхронизации?
Установите для offset значение latest. Задание будет потреблять только последние инкрементальные изменения и больше не будет выполнять полную инициализацию.