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

PostgreSQL

Doris предоставляет следующие способы загрузки данных из PostgreSQL:

  • Загрузка данных PostgreSQL через JDBC Catalog

Doris использует JDBC Catalog для представления PostgreSQL в виде внешнего Catalog, что позволяет выполнять SQL-запросы непосредственно к данным PostgreSQL. В сочетании с INSERT INTO или CREATE TABLE AS SELECT этот способ подходит для разовой миграции или периодической пакетной загрузки.

  • Непрерывная синхронизация данных PostgreSQL через Streaming Job

Doris использует Streaming Job для непрерывной полной и инкрементальной синхронизации данных из PostgreSQL в Doris. Благодаря интеграции возможностей чтения Flink CDC Doris поддерживает непрерывную работу задания, читает WAL из PostgreSQL и записывает данные в таблицы Doris с семантикой exactly-once. Поддерживаются два режима: SQL Mapping Sync и Auto Table Creation Sync. Доступно начиная с Doris 4.1.0.

  • Загрузка данных PostgreSQL через Flink Doris Connector

Используйте Flink Doris Connector вместе с Flink Postgres CDC для синхронизации в реальном времени. Этот способ подходит для сценариев, требующих дополнительной логики потоковой обработки во Flink. Коннектор также предоставляет инструмент для синхронизации всей базы данных одним нажатием. Подробнее см. в разделе Flink Doris Connector.

  • Загрузка данных PostgreSQL с помощью сторонних инструментов

Инструменты интеграции данных, такие как DataX, SeaTunnel и CloudCanal, также поддерживают синхронизацию данных из PostgreSQL в Doris.

В большинстве случаев для разовой миграции данных можно напрямую использовать JDBC Catalog. Если требуется непрерывная полная и инкрементальная синхронизация, рекомендуется Streaming Job.

Загрузка данных PostgreSQL через JDBC Catalog​

Используйте JDBC Catalog для представления PostgreSQL в виде внешнего Catalog, а затем загружайте данные с помощью INSERT INTO или CREATE TABLE AS SELECT. Подробное описание синтаксиса см. в разделе JDBC PostgreSQL Catalog.

Шаг 1: Подготовка данных в PostgreSQL​

CREATE TABLE public.students (
    id    INT PRIMARY KEY,
    name  VARCHAR(64),
    age   INT
);

INSERT INTO public.students VALUES (1, 'Emily', 25), (2, 'Bob', 30);

Шаг 2: Создание Catalog в Doris​

CREATE CATALOG pg_catalog PROPERTIES (
    "type"         = "jdbc",
    "user"         = "postgres",
    "password"     = "postgres",
    "jdbc_url"     = "jdbc:postgresql://127.0.0.1:5432/postgres",
    "driver_url"   = "postgresql-42.5.1.jar",
    "driver_class" = "org.postgresql.Driver"
);

Шаг 3: Создание целевой таблицы в Doris​

CREATE DATABASE IF NOT EXISTS doris_db;

CREATE TABLE doris_db.students (
    id    INT,
    name  VARCHAR(64),
    age   INT
)
UNIQUE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES ("replication_num" = "1");

Шаг 4: Загрузка данных с помощью INSERT INTO​

INSERT INTO doris_db.students
SELECT id, name, age FROM pg_catalog.postgres.public.students;

Если целевая таблица ещё не существует, можно также использовать CREATE TABLE AS SELECT, чтобы создать таблицу и загрузить данные за один шаг:

CREATE TABLE doris_db.students
PROPERTIES ("replication_num" = "1")
AS
SELECT * FROM pg_catalog.postgres.public.students;

Шаг 5: Проверка загруженных данных​

SELECT * FROM doris_db.students;
+----+-------+------+
| id | name  | age  |
+----+-------+------+
|  1 | Emily |   25 |
|  2 | Bob   |   30 |
+----+-------+------+

Непрерывная синхронизация данных PostgreSQL через Streaming Job​

Streaming Job непрерывно читает WAL PostgreSQL через Flink CDC и записывает данные в Doris. Поддерживаются два режима:

  • PostgreSQL CDC с автоматическим созданием таблиц: Doris автоматически создаёт целевые таблицы при первой синхронизации (используйте include_tables для синхронизации одной, нескольких или всех таблиц). Обеспечивает семантику at-least-once.
  • PostgreSQL CDC с сопоставлением через SQL: целевая таблица должна быть заранее создана в Doris. Поддерживает гибкое сопоставление столбцов, преобразование данных и семантику exactly-once.

Ограничения​

  1. Поддерживаются только таблицы с первичным ключом (Unique Key).
  2. Требуется привилегия Load. Для режима Auto Table Creation Sync также требуется привилегия Create, необходимая для автоматического создания целевых таблиц при первом запуске.
  3. Доступно начиная с Doris 4.1.0.

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

Перед отправкой задания Streaming Job на стороне PostgreSQL необходимо включить логическую репликацию (wal_level=logical), а пользователю предоставить соответствующие привилегии REPLICATION. Инструкции по настройке для конкретных сред см. в следующих разделах:

Пример использования: Auto Table Creation Sync​

Режим Auto Table Creation Sync использует синтаксис FROM POSTGRES ... TO DATABASE .... В качестве целевого объекта выступает база данных Doris, а целевые таблицы автоматически создаются при первой синхронизации.

Шаг 1: Подготовка данных в PostgreSQL​

CREATE TABLE public.students (
    id    INT PRIMARY KEY,
    name  VARCHAR(64),
    age   INT
);

INSERT INTO public.students VALUES (1, 'Emily', 25), (2, 'Bob', 30);

Шаг 2: Создание целевой базы данных в Doris​

Режим Auto Table Creation Sync не требует предварительного создания таблиц, но целевая база данных, в которой они будут размещены, должна существовать:

CREATE DATABASE IF NOT EXISTS doris_db;

Шаг 3: Создание задания Streaming Job​

В приведённом ниже примере параметр include_tables используется для синхронизации только таблицы students (для нескольких таблиц укажите их имена через запятую; для синхронизации всей базы данных оставьте значение пустым):

CREATE JOB pg_db_sync
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" = "students",
    "offset"         = "initial"
)
TO DATABASE doris_db (
    "table.create.properties.replication_num" = "1"  -- set to 1 in single-BE deployments
);

Шаг 4: Проверка состояния задания​

SELECT * FROM jobs("type"="insert") WHERE ExecuteType = "STREAMING";

Шаг 5: Просмотр автоматически созданных таблиц Doris и загруженных данных​

SHOW TABLES FROM doris_db;
SELECT * FROM doris_db.students;

Описание других распространённых операций и полный справочник параметров см. в разделе PostgreSQL CDC с автоматическим созданием таблиц.

Пример использования: SQL Mapping Sync​

Шаг 1: Подготовка данных в PostgreSQL​

CREATE TABLE public.students (
    id    INT PRIMARY KEY,
    name  VARCHAR(64),
    age   INT
);

INSERT INTO public.students VALUES (1, 'Emily', 25), (2, 'Bob', 30);

Шаг 2: Создание целевой таблицы в Doris​

Для режима SQL Mapping Sync целевая таблица должна быть создана заранее:

CREATE DATABASE IF NOT EXISTS doris_db;

CREATE TABLE doris_db.students (
    id    INT,
    name  VARCHAR(64),
    age   INT
)
UNIQUE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES ("replication_num" = "1");

Шаг 3: Создание задания Streaming Job​

Используйте CREATE STREAMING JOB с синтаксисом INSERT INTO ... SELECT * FROM cdc_stream(...):

CREATE JOB pg_students_sync
ON STREAMING
DO
INSERT INTO doris_db.students
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"        = "students",
    "offset"       = "initial"
);

Шаг 4: Проверка состояния задания​

SELECT * FROM jobs("type"="insert") WHERE ExecuteType = "STREAMING";

Шаг 5: Проверка загруженных данных​

SELECT * FROM doris_db.students;

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

Report issue