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

Синхронизация MySQL CDC с SQL-сопоставлением

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

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

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

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

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

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

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

ПунктОписание
Версия Doris4.1.5 и выше
Тип таблицыВ настоящее время в качестве целевой таблицы поддерживаются только таблицы с первичным ключом
Пользовательские привилегииТребуется привилегия 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. Операции с заданием​

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

Описание параметров​

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

Ниже приведены параметры источника данных MySQL, поддерживаемые CDC Stream TVF:

ПараметрЗначение по умолчаниюОписание
type-Тип источника данных. Укажите mysql
jdbc_url-Строка подключения JDBC к MySQL
driver_url-Путь к JAR-файлу драйвера JDBC. Поддерживаются имя файла, локальный абсолютный путь и URL HTTP. Подробности см. в Обзор JDBC Catalog
driver_class-Имя класса драйвера JDBC
user-Имя пользователя базы данных
password-Пароль базы данных
database-Имя базы данных
table-Имя синхронизируемой таблицы
offsetlatestlatest: только инкрементальная синхронизация; initial: полная + инкрементальная синхронизация
snapshot_split_size40960Размер split (в строках). При полной синхронизации таблица делится на несколько split для синхронизации
snapshot_parallelism1Параллелизм на этапе полной синхронизации, то есть максимальное число split, планируемых для одной Task

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

ПараметрЗначение по умолчаниюОписание
session.*NoneВсе переменные сеанса можно настроить в job_properties. Сведения о переменных импорта см. в Insert Into Select

Дополнительные общие параметры (например, max_interval) см. в разделе Обзор непрерывного импорта.

FAQ​

Q1: В чём разница между SQL Mapping Sync и синхронизацией с автоматическим созданием таблиц?

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

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

В настоящее время в качестве целевой таблицы поддерживаются только таблицы с первичным ключом.

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

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

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

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

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