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

CDC_STREAM

Описание​

Табличная функция (TVF) CDC Stream позволяет читать изменения данных из реляционных баз данных (таких как MySQL и PostgreSQL) через CDC. Благодаря интеграции возможностей чтения Flink CDC она поддерживает полную и инкрементальную синхронизацию данных.

Обычно она используется с CREATE JOB ON STREAMING для непрерывной синхронизации данных одной таблицы с сопоставлением с помощью SQL. Подробности см. в разделах CDC из MySQL с сопоставлением с помощью SQL и CDC из PostgreSQL с сопоставлением с помощью SQL.

Синтаксис​

cdc_stream(
    "type" = "<source_type>",
    "jdbc_url" = "<jdbc_url>",
    "driver_url" = "<driver_url>",
    "driver_class" = "<driver_class>",
    "user" = "<user>",
    "password" = "<password>",
    "database" = "<database>",
    "table" = "<table>"
    [, "<optional_property_key>" = "<optional_property_value>" [, ...] ]
)

Обязательные параметры​

ПараметрОписание
typeТип источника данных; в настоящее время поддерживаются mysql и postgres
jdbc_urlСтрока подключения JDBC, например jdbc:mysql://127.0.0.1:3306 или jdbc:postgresql://127.0.0.1:5432/postgres
driver_urlПуть к jar-файлу драйвера JDBC
driver_classИмя класса драйвера JDBC. com.mysql.cj.jdbc.Driver для MySQL, org.postgresql.Driver для PostgreSQL
userИмя пользователя базы данных
passwordПароль базы данных
databaseИмя базы данных
tableИмя таблицы для синхронизации

Необязательные параметры​

ПараметрПо умолчаниюОписание
schema-Имя схемы; обязательно для PostgreSQL
offsetlatestlatest: только инкрементальная синхронизация; initial: полная + инкрементальная синхронизация; также поддерживаются snapshot, earliest (только для MySQL) или позиция binlog/LSN в формате JSON
snapshot_split_size40960 (8096 до 4.1.4)Размер фрагмента (в строках). При полной синхронизации таблица разбивается на несколько фрагментов
snapshot_parallelism1Степень параллелизма на этапе полной синхронизации, то есть максимальное число фрагментов на задачу

Примечания​

  1. CDC Stream TVF обычно используется с CREATE JOB ON STREAMING и не рекомендуется для обычных запросов.
  2. При использовании источника типа MySQL необходимо включить Binlog в MySQL (binlog_format=ROW).
  3. При использовании источника типа PostgreSQL необходимо включить логическую репликацию (wal_level=logical).
  4. Поддерживается семантика exactly-once.

Начиная с 4.1.5, при использовании со Streaming Job операции CDC проверяют права доступа в Doris от имени создателя задания, а для задач, завершившихся с ошибкой, не применяется обычный механизм автоматического повторного запуска. См. раздел Учетная запись и обработка сбоев при непрерывной загрузке.

Примеры​

  • Непрерывная синхронизация одной таблицы из MySQL

    CREATE JOB mysql_cdc_job
    ON STREAMING
    DO
    INSERT INTO db1.target_table
    SELECT * FROM cdc_stream(
        "type" = "mysql",
        "jdbc_url" = "jdbc:mysql://127.0.0.1:3306",
        "driver_url" = "mysql-connector-java-8.0.25.jar",
        "driver_class" = "com.mysql.cj.jdbc.Driver",
        "user" = "root",
        "password" = "123456",
        "database" = "source_db",
        "table" = "source_table",
        "offset" = "initial"
    )

  • Непрерывная синхронизация одной таблицы из PostgreSQL

    CREATE JOB pg_cdc_job
    ON STREAMING
    DO
    INSERT INTO db1.target_table
    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"
    )

  • Только инкрементальная синхронизация (без полного снимка)

    CREATE JOB mysql_incremental_job
    ON STREAMING
    DO
    INSERT INTO db1.target_table
    SELECT * FROM cdc_stream(
        "type" = "mysql",
        "jdbc_url" = "jdbc:mysql://127.0.0.1:3306",
        "driver_url" = "mysql-connector-java-8.0.25.jar",
        "driver_class" = "com.mysql.cj.jdbc.Driver",
        "user" = "root",
        "password" = "123456",
        "database" = "source_db",
        "table" = "source_table",
        "offset" = "latest"
    )

  • Сопоставление столбцов и преобразование данных

    CREATE JOB mysql_cdc_transform_job
    ON STREAMING
    DO
    INSERT INTO db1.target_table (id, name, age)
    SELECT id, name, cast(age as INT) as age
    FROM cdc_stream(
        "type" = "mysql",
        "jdbc_url" = "jdbc:mysql://127.0.0.1:3306",
        "driver_url" = "mysql-connector-java-8.0.25.jar",
        "driver_class" = "com.mysql.cj.jdbc.Driver",
        "user" = "root",
        "password" = "123456",
        "database" = "source_db",
        "table" = "source_table",
        "offset" = "initial"
    )
Report issue