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

PostgreSQL

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

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

Doris подключает PostgreSQL как внешний Catalog с помощью JDBC 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​

Подключите PostgreSQL как внешний Catalog с помощью JDBC 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