CDC MySQL с автоматическим созданием таблиц

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

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

Синхронизация с автоматическим созданием таблиц реализуется через встроенный DDL FROM MYSQL (...) TO DATABASE (...). Целью является база данных Doris, а Doris автоматически создает соответствующие целевые таблицы на основе схем исходных таблиц. С помощью include_tables можно указать, синхронизировать ли одну, несколько или все таблицы. При первой синхронизации Doris автоматически создает целевые таблицы с первичными ключами и сохраняет первичные ключи, согласованные с исходными. Этот режим подходит для зеркальной репликации, когда обработка SQL не требуется, а схема целевой таблицы должна автоматически следовать схеме источника.

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

Подходящие сценарии

• Требуется зеркально реплицировать группу таблиц (или всю базу данных) из MySQL в Doris.
• Нужно автоматически создать схему целевых таблиц и первичные ключи на основе источника без ручного создания таблиц.
• Требуется только простое переименование целевых таблиц или исключение столбцов, без SQL-выражений, фильтрации строк и преобразования данных.
• Нужно поддерживать как первоначальную полную инициализацию, так и непрерывное получение инкрементальных изменений.

Возможности и ограничения

Пункт	Описание
Семантика согласованности	В настоящее время гарантируется только семантика at-least-once
Тип таблицы	Можно синхронизировать только исходные таблицы с первичными ключами; автоматически создаваемые целевые таблицы используют модель Unique Key
Разрешения	Требуется разрешение Load; если целевая таблица не существует, также требуется разрешение Create
Автоматическое создание таблиц	Если целевая таблица уже существует, ее создание пропускается; схему таблицы можно настроить по необходимости
Обработка данных	Поддерживаются переименование целевых таблиц и исключение столбцов без ключей; SQL-выражения, фильтрация строк и преобразование данных не поддерживаются

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

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

1. Кластер Doris развернут, и у вас есть разрешение Load (для сценариев автоматического создания таблиц также требуется разрешение Create).
2. Загружен JDBC-драйвер jar, совместимый с версией MySQL, на который можно ссылаться по имени файла, локальному абсолютному пути или HTTP-адресу. Подробности см. в разделе Обзор JDBC Catalog.
3. В MySQL включен binlog, а учетная запись имеет разрешение на его чтение.
4. Область синхронизации определена: синхронизируется вся база данных или только таблицы, указанные в include_tables.

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

Шаг 1. Создайте задание загрузки

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

CREATE JOB multi_table_sync
ON STREAMING
FROM 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" = "test",
        "include_tables" = "user_info,order_info",
        "offset" = "initial"
)
TO DATABASE target_test_db (
    "table.create.properties.replication_num" = "1"  -- Set to 1 for single-BE deployment
)

Основные моменты приведенного примера:

• database задает исходную базу данных MySQL. include_tables ограничивает набор синхронизируемых таблиц; несколько таблиц разделяются запятыми, а пустое значение означает синхронизацию всей базы данных.
• offset = "initial" означает, что сначала выполняется полная инициализация, а затем начинается инкрементальная синхронизация. Чтобы синхронизировать только инкрементальные данные, измените значение на latest.
• TO DATABASE target_test_db задает целевую базу данных Doris. Используйте table.create.properties.*, чтобы управлять свойствами автоматически создаваемых таблиц, например количеством реплик.

Шаг 2. Проверьте состояние загрузки

Запросите состояние выполнения Streaming Job с помощью табличной функции jobs("type"="insert"):

select * from jobs("type"="insert") where ExecuteType = "STREAMING"
             Id: 1765332859199
             Name: mysql_db_sync
          Definer: root
      ExecuteType: STREAMING
RecurringStrategy: \N
           Status: RUNNING
       ExecuteSql: FROM MYSQL('include_tables'='user_info','database'='test','driver_class'='com.mysql.cj.jdbc.Driver','driver_url'='mysql-connector-java-8.0.25.jar','offset'='initial','jdbc_url'='jdbc:mysql://127.0.0.1:3306','user'='root' ) TO DATABASE target_test_db ('table.create.properties.replication_num'='1')
       CreateTime: 2025-12-10 10:19:35
 SucceedTaskCount: 1
  FailedTaskCount: 0
CanceledTaskCount: 0
          Comment:
       Properties: \N
    CurrentOffset: {"ts_sec":"1765284495","file":"binlog.000002","pos":"9350","kind":"SPECIFIC","splitId":"binlog-split","row":"1","event":"2","server_id":"1"}
        EndOffset: \N
    LoadStatistic: {"scannedRows":24,"loadBytes":1146,"fileNumber":0,"fileSize":0}
         ErrorMsg: \N

Обратите внимание на следующие поля:

• Status: общее состояние задания, например RUNNING.
• CurrentOffset: текущая позиция binlog, которую уже обработало задание; полезна для отслеживания хода синхронизации.
• LoadStatistic: статистика загрузки, например количество просканированных строк и записанных байтов.
• ErrorMsg: сообщения об ошибках, основной источник информации для устранения неполадок задания.

Шаг 3. Измените задание загрузки

Если требуется обновить сведения о подключении к источнику данных, например сменить учетные данные, используйте ALTER JOB:

ALTER JOB <job_name>
FROM MYSQL (
    "user" = "root",
    "password" = "123456"
)
TO DATABASE target_test_db

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

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

Источник MySQL (FROM MYSQL) поддерживает следующие параметры. Сведения о подключении, имя базы данных и сведения о драйвере обязательны.

Параметр	Обязательный	По умолчанию	Описание
jdbc_url	Да	-	Строка подключения MySQL JDBC.
driver_url	Да	-	Путь к jar-файлу JDBC-драйвера. Поддерживаются имя файла, локальный абсолютный путь или HTTP-URL. Подробности см. в разделе Обзор JDBC Catalog.
driver_class	Да	-	Имя класса JDBC-драйвера, например com.mysql.cj.jdbc.Driver.
user	Да	-	Имя пользователя базы данных.
password	Да	-	Пароль базы данных.
database	Да	-	Имя базы данных MySQL.
include_tables	Нет	-	Имена синхронизируемых таблиц через запятую. Если параметр не задан, синхронизируются все таблицы базы данных.
exclude_tables	Нет	-	Имена таблиц, которые не нужно синхронизировать, через запятую. Параметр действует только если include_tables не задан.
table.<table_name>.target_table	Нет	Имя исходной таблицы	Задает имя целевой таблицы Doris для исходной таблицы. <table_name> — имя исходной таблицы.
table.<table_name>.exclude_columns	Нет	-	Указывает столбцы исходной таблицы, которые не нужно синхронизировать, через запятую. Столбцы должны существовать и не могут включать столбцы первичного ключа.
offset	Нет	latest	Начальная позиция. initial: полная и инкрементальная синхронизация; earliest: начать с самой ранней доступной позиции Binlog; latest: синхронизировать только изменения после запуска задания. Можно также указать точную позицию в формате JSON, например {"file":"binlog.000001","pos":"154"} или {"gtids":"<gtid_set>"}. Также поддерживается snapshot, предназначенный только для полной синхронизации.
snapshot_split_size	Нет	40960	Размер split в строках. Во время полной синхронизации таблица делится на несколько split. Должно быть положительным целым числом.
snapshot_parallelism	Нет	1	Степень параллелизма этапа полной синхронизации, то есть максимальное число split, планируемых одной Task одновременно. Должно быть положительным целым числом.
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_tables и exclude_tables, приоритет имеет include_tables. Следующий фрагмент параметров FROM MYSQL синхронизирует orders и customers, записывает orders в ods_orders, исключает столбец без ключа internal_note и включает проверку CA:

"include_tables" = "orders,customers",
"table.orders.target_table" = "ods_orders",
"table.orders.exclude_columns" = "internal_note",
"server_id" = "5400-5403",
"ssl_mode" = "verify-ca",
"ssl_rootcert" = "FILE:ca.pem"

Справочник

Команда загрузки

Синтаксис создания задания синхронизации с автоматическим созданием таблиц имеет следующий вид:

CREATE JOB <job_name>
[job_properties]
ON STREAMING
[ COMMENT <comment> ]
FROM MYSQL (
    [source_properties]
)
TO DATABASE <target_db> (
    [target_properties]
)

Описание каждого модуля:

Модуль	Описание
job_name	Имя задания
job_properties	Используется для задания общих параметров загрузки Job
comment	Используется для добавления примечания с описанием Job
source_properties	Параметры, относящиеся к источнику MySQL
target_properties	Параметры, относящиеся к целевой базе данных Doris

job_properties поддерживает max_interval и compute_group. Подробности см. в разделе Общие параметры конфигурации загрузки задания. В режиме автоматического создания таблиц session.* не используется.

Параметры конфигурации целевой базы данных Doris

TO DATABASE поддерживает следующие необязательные параметры:

Параметр	По умолчанию	Описание
table.create.properties.*	-	Добавляет свойства таблицы при создании таблицы Doris, например table.create.properties.replication_num.
load.strict_mode	false	Определяет, следует ли включить строгий режим для записи через Stream Load. Допустимые значения: true и false.
load.max_filter_ratio	0	Максимально допустимая доля фильтрации в окне выборки, в диапазоне [0, 1]. Значение 0 означает, что ошибочные строки нельзя отфильтровать. Окно выборки составляет max_interval * 10 секунд. Задание приостанавливается, если отношение числа ошибочных строк к общему числу строк в окне превышает это значение.

FAQ

Q1. Поддерживает ли синхронизация с автоматическим созданием таблиц таблицы без первичного ключа?

Нет. В настоящее время синхронизация с автоматическим созданием таблиц поддерживает только таблицы с первичным ключом (Unique Key). При первой синхронизации Doris автоматически создает целевые таблицы с первичными ключами на основе первичных ключей источника.

Q2. Будет ли перезаписана уже существующая целевая таблица?

Нет. На этапе автоматического создания таблиц создание пропускается, если целевая таблица уже существует. Схему таблицы можно настроить в соответствии с конкретным сценарием.

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

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

Q4. Как выбрать между синхронизацией с автоматическим созданием таблиц и синхронизацией с сопоставлением SQL?

• Для зеркальной репликации, автоматического создания таблиц и сохранения схемы таблиц согласованной с источником используйте синхронизацию с автоматическим созданием таблиц.
• Если требуется только переименовать целевые таблицы или исключить столбцы без ключей, используйте синхронизацию с автоматическим созданием таблиц.
• Если нужны SQL-выражения, фильтрация строк или преобразование данных, используйте CDC MySQL с сопоставлением SQL.

Q5. Что делать, если создание таблицы завершается ошибкой в развертывании с одним BE?

Явно укажите число реплик, равное 1, в TO DATABASE: "table.create.properties.replication_num" = "1".

Связанные документы

• Обзор непрерывной загрузки
• CDC MySQL с сопоставлением SQL
• CREATE STREAMING JOB
• Обзор JDBC Catalog
• Flink CDC