Последнее обновление 3 нояб. 2026 г.

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

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

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

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

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

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

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

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

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

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

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

Кластер Doris развёрнут, и у вас есть разрешение Load (для сценариев автоматического создания таблиц также требуется разрешение Create).
Загружен JAR-файл JDBC-драйвера, совместимый с версией MySQL, и на него можно ссылаться по имени файла, локальному абсолютному пути или HTTP-адресу. Подробности см. в разделе Обзор JDBC Catalog.
В MySQL включён binlog, а учётная запись имеет разрешение на чтение binlog.
Определена область синхронизации: вся база данных или только таблицы, указанные в 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-адрес. Подробности см. в разделе Обзор JDBC Catalog
driver_class	-	Имя класса JDBC-драйвера
user	-	Имя пользователя базы данных
password	-	Пароль базы данных
database	-	Имя базы данных
include_tables	-	Имена синхронизируемых таблиц, разделённые запятыми. Если оставить поле пустым, по умолчанию синхронизируются все таблицы
offset	latest	`latest`: только инкрементальная синхронизация; `initial`: полная + инкрементальная синхронизация
snapshot_split_size	40960	Размер split в строках. Во время полной синхронизации таблица делится на несколько split для синхронизации
snapshot_parallelism	1	Параллелизм на этапе полной синхронизации, то есть максимальное число split, которое одна Task может запланировать одновременно

Справочное руководство

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

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

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

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

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

Параметр	Значение по умолчанию	Описание
table.create.properties.*	-	Задаёт свойства таблицы при её создании, например `replication_num`
load.strict_mode	-	Включать ли строгий режим. По умолчанию отключён
load.max_filter_ratio	-	Максимальный коэффициент фильтрации, допустимый в окне выборки. Должен быть больше или равен 0 и меньше или равен 1. Значение по умолчанию — 0, что означает нулевую допустимость. Окно выборки равно `max_interval * 10`. Если отношение ошибочных строк к общему числу строк в окне выборки превышает `load.max_filter_ratio`, routine job приостанавливается, и требуется ручное вмешательство для проверки качества данных.

FAQ

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

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

Вопрос 2. Будет ли целевая таблица перезаписана, если она уже существует?

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

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

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

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

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

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

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

Устранение неполадок

Симптом	Возможная причина	Диагностика и решение
Создание задания завершается ошибкой доступа	У текущего пользователя отсутствует разрешение Load или Create	Предоставьте выполняющей учётной записи разрешение Load; для автоматического создания таблиц также требуется Create
Полная синхронизация выполняется медленно	Слишком малое значение `snapshot_parallelism` или слишком большие split	Увеличьте `snapshot_parallelism` в разумных пределах; скорректируйте `snapshot_split_size` с учётом числа строк в таблице
Задание приостановлено из-за проблем с качеством данных	Доля ошибочных строк превышает `load.max_filter_ratio`	Проверьте поле `ErrorMsg`, чтобы найти некорректные данные; измените `load.strict_mode` или `load.max_filter_ratio`
Ошибка числа реплик при автоматическом создании таблицы	Развёртывание с одним BE, а число реплик по умолчанию равно 3	Установите `table.create.properties.replication_num = 1`
Инкрементальные данные не поступают в Doris	В MySQL не включён binlog или у учётной записи нет разрешения на binlog	Включите binlog в MySQL и предоставьте учётной записи синхронизации разрешения `REPLICATION SLAVE` и `REPLICATION CLIENT`

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

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

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

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

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

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

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

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

Справочное руководство​

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

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

FAQ​

Устранение неполадок​

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