Последнее обновление 3 нояб. 2026 г.

PostgreSQL CDC с сопоставлением SQL

Экспериментальная функция

Непрерывная загрузка Doris является экспериментальной функцией и поддерживается начиная с Doris 4.1.5.

Синхронизация с сопоставлением SQL реализуется посредством Job и CDC Stream TVF. Целевым объектом является уже существующая таблица Doris (INSERT INTO tbl SELECT * FROM cdc_stream(...)). SQL Doris предоставляет возможности для сопоставления столбцов, фильтрации и преобразования данных. Этот подход подходит для сценариев синхронизации в реальном времени, где данные необходимо обработать перед записью.

Интегрируя возможность чтения Flink CDC, Doris считывает журналы изменений (WAL) из PostgreSQL и выполняет полную и инкрементальную синхронизацию исходной таблицы с целевой. Если требуется, чтобы Doris автоматически создавал целевые таблицы и синхронизировал группу таблиц по каждой базе данных, см. PostgreSQL CDC с автоматическим созданием таблиц.

Сценарии использования

Приём данных одной таблицы в хранилище в реальном времени, когда перед записью требуется удаление ненужных столбцов, фильтрация или преобразование на основе выражений.
Целевая таблица Doris уже существует, и требуется явно управлять сопоставлением полей с помощью SQL.
Высокие требования к согласованности, требующие семантики exactly-once.

Предварительные условия

Перед созданием задания убедитесь в следующем:

На стороне PostgreSQL включена логическая репликация. Подробности см. в руководстве по настройке.
Текущий пользователь имеет привилегию Load.
Исходная таблица является таблицей с первичным ключом (в настоящее время поддерживается синхронизация только таких таблиц).
Целевая таблица Unique Key заранее создана в Doris, а её схема совместима с SQL сопоставления.

Примечания

Поддерживается семантика exactly-once.
В настоящее время поддерживается только синхронизация таблиц с первичным ключом.
Требуется привилегия Load.
На стороне PostgreSQL должна быть включена логическая репликация. См. руководство по настройке.

Быстрый запуск

Общий процесс состоит из двух шагов: создать задание непрерывной загрузки, затем проверить его состояние.

Шаг 1. Создание задания загрузки

Используйте CREATE STREAMING JOB, чтобы создать задание непрерывной загрузки, синхронизирующее public.source_table из PostgreSQL с db1.tbl1 в Doris:

CREATE JOB pg_single_sync
ON STREAMING
DO
INSERT INTO db1.tbl1
SELECT * FROM cdc_stream(
    "type" = "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",
    "table" = "source_table",
    "offset" = "initial"
)

Шаг 2. Проверка состояния загрузки

Используйте табличную функцию jobs, чтобы проверить состояние выполнения заданий типа Streaming:

select * from jobs("type"="insert") where ExecuteType = "STREAMING";

Другие распространённые операции (приостановка, возобновление, удаление, просмотр задач и т. д.) описаны в Обзоре непрерывной загрузки.

Справочник параметров

Параметры источника данных

Параметры источника данных задаются через TVF cdc_stream(...). Они описывают синхронизируемую таблицу PostgreSQL и поведение чтения.

Параметр	Обязательный	Значение по умолчанию	Описание
`type`	Да	-	Тип источника данных. Укажите `postgres`.
`jdbc_url`	Да	-	Строка подключения PostgreSQL JDBC.
`driver_url`	Да	-	Путь к jar-файлу драйвера JDBC. Поддерживаются имя файла, локальный абсолютный путь или URL HTTP. Подробности см. в Обзоре JDBC Catalog.
`driver_class`	Да	-	Имя класса драйвера JDBC, например `org.postgresql.Driver`.
`user`	Да	-	Имя пользователя базы данных.
`password`	Да	-	Пароль базы данных.
`database`	Нет	Имя базы данных в JDBC URL	Имя базы данных PostgreSQL. При явной установке это значение заменяет имя базы данных в JDBC URL. После кодирования UTF-8 длина не должна превышать 63 байта.
`schema`	Да	-	Имя схемы.
`table`	Да	-	Имя синхронизируемой таблицы. Каждое задание SQL Mapping поддерживает одну исходную таблицу.
`offset`	Да	-	Начальная позиция. `initial`: полная и инкрементальная синхронизация; `snapshot`: только полная синхронизация; `latest`: синхронизация только изменений после запуска задания. Можно также указать точную позицию JSON, например `{"lsn":"12345678"}`. PostgreSQL не поддерживает `earliest`.
`snapshot_split_size`	Нет	`40960`	Размер split в строках. При полной синхронизации таблица делится на несколько split. Должно быть положительным целым числом.
`snapshot_parallelism`	Нет	`1`	Параллелизм этапа полной синхронизации, то есть максимальное число split, планируемых Task одновременно. Должно быть положительным целым числом.
`skip_snapshot_backfill`	Нет	`false`	Нужно ли пропускать заполнение WAL во время снимка. При значении `true` используется семантика at-least-once.
`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_delete_sign`	Нет	`false`	Нужно ли TVF дополнительно выводить столбец `__DORIS_DELETE_SIGN__`. Установите `true`, чтобы синхронизировать операции DELETE из источника как удаления в таблице Doris с первичным ключом.

При включении include_delete_sign целевая таблица должна быть таблицей Unique Key с Merge-on-Write, а __DORIS_DELETE_SIGN__ необходимо явно сопоставить и в списке целевых столбцов INSERT, и в списке SELECT:

CREATE JOB pg_cdc_with_delete
ON STREAMING
DO
INSERT INTO db1.target_table (id, value, __DORIS_DELETE_SIGN__)
SELECT id, value, __DORIS_DELETE_SIGN__
FROM cdc_stream(
    "type" = "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",
    "schema" = "public",
    "table" = "source_table",
    "offset" = "initial",
    "include_delete_sign" = "true"
);

При настройке собственного slot_name или publication_name подготовьте соответствующий ресурс до создания Job. После удаления Job Doris сохраняет ресурсы, предоставленные пользователем.

Параметры конфигурации Job

Следующие параметры задаются через CREATE JOB ... PROPERTIES (...):

Параметр	Значение по умолчанию	Описание
`max_interval`	`10`	Интервал планирования простоя в секундах, когда новые данные из источника отсутствуют. Должен быть целым числом не меньше 1.
`compute_group`	Текущая сессия или вычислительная группа пользователя по умолчанию	Поддерживается только в режиме разделения вычислений и хранения. Указывает вычислительную группу, в которой выполняется задание. Пользователь должен иметь привилегию USAGE для этой вычислительной группы.
`session.<variable_name>`	Значение соответствующей переменной сеанса по умолчанию	Задаёт переменную сеанса для задачи INSERT. Переменные загрузки см. в Insert Into Select.

Описание свойства Job offset, используемого для сброса позиции CDC через ALTER JOB, и все ограничения см. в Обзоре непрерывной загрузки. Начальную позицию для нового задания необходимо задать в cdc_stream(...).

Сценарии использования​

Предварительные условия​

Примечания​

Быстрый запуск​

Шаг 1. Создание задания загрузки​

Шаг 2. Проверка состояния загрузки​

Справочник параметров​

Параметры источника данных​

Параметры конфигурации Job​