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

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

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

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

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

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

Сценарии применения​

  • Непрерывная синхронизация одной таблицы MySQL с Doris, когда схема целевой таблицы уже спланирована
  • Во время синхронизации требуется исключение столбцов, сопоставление столбцов, переименование полей или преобразование данных
  • Интеграция данных в реальном времени, требующая семантики exactly-once на всём пути обработки

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

ПунктОписание
Версия Doris4.1.5 и выше
Тип таблицыИсходная таблица выше по потоку должна иметь первичный ключ; целевая таблица Doris должна использовать модель Unique Key
Права пользователяТребуется привилегия Load
Конфигурация MySQLНа стороне MySQL должен быть включён Binlog. См. Руководство по настройке
Гарантия семантикиПоддерживается семантика exactly-once

Быстрый старт​

Ниже приведён минимальный работоспособный пример полного процесса: создание задания и проверка его состояния.

Шаг 1. Создание задания импорта​

Используйте CREATE STREAMING JOB, чтобы создать задание непрерывного импорта:

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

Шаг 2. Проверка статуса импорта​

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

Шаг 3. Операции с заданием​

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

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

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

CDC Stream TVF поддерживает следующие параметры источника данных MySQL.

ПараметрОбязательныйПо умолчаниюОписание
typeДа-Тип источника данных. Укажите mysql.
jdbc_urlДа-Строка подключения MySQL JDBC.
driver_urlДа-Путь к JAR-файлу драйвера JDBC. Поддерживаются имя файла, локальный абсолютный путь или URL HTTP. Подробнее см. Обзор JDBC Catalog.
driver_classДа-Имя класса драйвера JDBC, например com.mysql.cj.jdbc.Driver.
userДа-Имя пользователя базы данных.
passwordДа-Пароль базы данных.
databaseДа-Имя базы данных MySQL.
tableДа-Имя таблицы для синхронизации. Каждое задание SQL Mapping поддерживает одну исходную таблицу.
offsetДа-Начальная позиция. initial: полная и инкрементальная синхронизация; snapshot: только полная синхронизация; earliest: запуск с самой ранней доступной позиции Binlog; latest: синхронизация только изменений после запуска задания. Также можно указать точную позицию JSON, например {"file":"binlog.000001","pos":"154"} или {"gtids":"<gtid_set>"}.
snapshot_split_sizeНет40960Размер split в строках. Во время полной синхронизации таблица делится на несколько split. Должно быть положительным целым числом.
snapshot_parallelismНет1Параллелизм этапа полной синхронизации, то есть максимальное число split, планируемых Task одновременно. Должно быть положительным целым числом.
skip_snapshot_backfillНетfalseСледует ли пропускать обратное заполнение Binlog во время снимка. При значении true используется семантика at-least-once.
server_idНетАвтоматически генерируетсяИдентификатор сервера ридера MySQL CDC. Поддерживается одно значение, например 5400, или замкнутый диапазон, например 5400-5408. Ширина диапазона должна быть больше или равна snapshot_parallelism.
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 mysql_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" = "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",
    "include_delete_sign" = "true"
);

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

Установите следующие параметры через CREATE JOB ... PROPERTIES (...):

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

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

FAQ​

Q1: Чем отличается синхронизация с сопоставлением SQL от синхронизации с автоматическим созданием таблиц?

  • Синхронизация с сопоставлением SQL: целевая таблица Doris должна быть создана заранее. Поддерживаются сопоставление столбцов и преобразование данных; этот вариант подходит для сценариев детальной обработки.
  • Синхронизация с автоматическим созданием таблиц: Doris автоматически создаёт нижестоящие таблицы и синхронизирует их целиком по базам данных. Подробнее см. Синхронизация MySQL CDC с автоматическим созданием таблиц.

Q2: Поддерживаются ли таблицы без первичного ключа в качестве целевой таблицы?

Исходная таблица выше по потоку должна иметь первичный ключ, а целевая таблица Doris в настоящее время должна использовать модель Unique Key.

Q3: Как синхронизировать только инкрементальные данные без исторических полных данных?

Установите для параметра offset значение latest. Задание пропустит этап полной синхронизации и будет синхронизировать только инкрементальные данные Binlog.

Q4: Как оптимизировать работу, если полная синхронизация выполняется слишком медленно?

Для повышения пропускной способности на этапе полной синхронизации можно изменить следующие два параметра:

  • snapshot_split_size: увеличьте число строк в одном split.
  • snapshot_parallelism: увеличьте параллелизм split, планируемых одним Task.

При изменении snapshot_parallelism, если также настроен диапазон server_id, убедитесь, что количество идентификаторов в диапазоне не меньше значения параллелизма.

Report issue