Перейти к содержимому
Последнее обновление

Синхронизация CDC PostgreSQL с сопоставлением SQL

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

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

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

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

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

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

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

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

  1. На стороне PostgreSQL включена логическая репликация. Подробности см. в Руководстве по настройке.
  2. Текущий пользователь имеет привилегию Load.
  3. Исходная таблица является таблицей с первичным ключом (в настоящее время поддерживается синхронизация только таблиц с первичным ключом).
  4. Целевая таблица заранее создана в 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.
user-Имя пользователя базы данных.
password-Пароль базы данных.
database-Имя базы данных.
schema-Имя схемы.
table-Имя синхронизируемой таблицы.
offsetlatestlatest: только инкрементальная синхронизация. initial: полная + инкрементальная синхронизация.
snapshot_split_size40960Размер split в строках. При полной синхронизации таблица делится на несколько split для синхронизации.
snapshot_parallelism1Параллелизм этапа полной синхронизации, то есть максимальное количество split, планируемых для одного Task.

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

Параметры конфигурации загрузки задаются через job_properties задания. Они управляют поведением загрузки и переменными сеанса.

ПараметрПо умолчаниюОписание
session.*NoneВсе переменные сеанса можно настроить в job_properties. Переменные, связанные с загрузкой, см. в Insert Into Select.

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

Report issue