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

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

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

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

Синхронизация OceanBase CDC с автоматическим созданием таблиц использует FROM OCEANBASE (...) TO DATABASE (...). Doris считывает полные и инкрементальные данные через сервис OBBinlog и автоматически создает таблицы Doris Unique Key на основе схем источника. Используйте include_tables, чтобы синхронизировать одну, несколько или все таблицы базы данных.

Непрерывная загрузка OceanBase CDC в настоящее время поддерживает только режим совместимости OceanBase с MySQL и Auto Table Creation Sync. Режим совместимости с Oracle и SQL Mapping Sync не поддерживаются.

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

  • Зеркалировать группу таблиц OceanBase или всю базу данных в Doris.
  • Разрешить Doris создавать целевые таблицы и первичные ключи на основе схем источника.
  • Выполнить первоначальную полную загрузку, а затем непрерывно синхронизировать инкрементальные изменения.
  • Переименовывать целевые таблицы или исключать столбцы без SQL-выражений, фильтров строк и преобразований данных.

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

ПунктОписание
Режим совместимостиТолько режим совместимости OceanBase с MySQL
Способ синхронизацииТолько Auto Table Creation Sync; SQL Mapping Sync не поддерживается
Семантика согласованностиAt-least-once
Тип таблицыТаблицы источника должны иметь первичные ключи; автоматически созданные таблицы Doris используют модель Unique Key
РазрешенияТребуется разрешение Load; если целевая таблица отсутствует, также требуется разрешение Create
Schema ChangeСинхронизируются ADD COLUMN и DROP COLUMN; см. Синхронизация изменений схемы
Типы данныхИспользуется преобразование типов, совместимое с MySQL; см. Сопоставление типов данных

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

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

  1. OceanBase работает в режиме совместимости с MySQL. При создании задания Doris проверяет режим, выполняя SHOW VARIABLES LIKE 'ob_compatibility_mode'.
  2. Сервис OBBinlog развернут и доступен. Выполнение SHOW MASTER STATUS через конечную точку его протокола MySQL возвращает допустимые файл binlog и позицию.
  3. Doris FE, BE и CDC Client могут обращаться к сервису OBBinlog.
  4. Доступен совместимый драйвер MySQL Connector/J, а jdbc_url начинается с jdbc:mysql://.
  5. Учетная запись источника может читать таблицы, схемы и binlog в области синхронизации.
  6. Каждая синхронизируемая исходная таблица имеет первичный ключ.

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

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

Следующий пример синхронизирует таблицы users и orders в source_db:

CREATE JOB oceanbase_db_sync
ON STREAMING
FROM OCEANBASE (
"jdbc_url" = "jdbc:mysql://127.0.0.1:2883",
"driver_url" = "mysql-connector-j-8.4.0.jar",
"driver_class" = "com.mysql.cj.jdbc.Driver",
"user" = "root@test",
"password" = "123456",
"database" = "source_db",
"include_tables" = "users,orders",
"offset" = "initial"
)
TO DATABASE target_db (
"table.create.properties.replication_num" = "1"
);
  • jdbc_url указывает на порт протокола MySQL сервиса OBBinlog. Используйте порт, настроенный в вашем развертывании.
  • offset = "initial" сначала загружает существующие строки, а затем непрерывно синхронизирует изменения binlog.
  • Если include_tables не указан, синхронизируются все подходящие таблицы в database.
  • Для развертывания с одним BE установите replication_num равным 1. В рабочей среде задайте количество реплик в соответствии с размером кластера.

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

SELECT Name, Status, CurrentOffset, ErrorMsg
FROM jobs("type" = "insert")
WHERE Name = "oceanbase_db_sync";

При нормальной работе задания значение Status равно RUNNING:

+-------------------+---------+-----------------------------------------------+----------+
| Name | Status | CurrentOffset | ErrorMsg |
+-------------------+---------+-----------------------------------------------+----------+
| oceanbase_db_sync | RUNNING | {"file":"binlog.000001","pos":"154", ...} | NULL |
+-------------------+---------+-----------------------------------------------+----------+

Шаг 3. Проверка инкрементальной синхронизации​

После выполнения INSERT, UPDATE или DELETE в исходной таблице OceanBase запросите соответствующую таблицу Doris и убедитесь, что изменение синхронизировано:

SELECT * FROM target_db.users ORDER BY id;

Операции приостановки, возобновления, удаления и запроса Task см. в разделе Обзор непрерывной загрузки.

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

ПараметрОбязательныйЗначение по умолчаниюОписание
jdbc_urlДа-Строка JDBC-подключения к сервису OBBinlog. Она должна начинаться с jdbc:mysql://.
driver_urlДа-Путь к jar-файлу драйвера MySQL Connector/J. Поддерживаются имя файла, локальный абсолютный путь или URL HTTP.
driver_classДа-Класс драйвера MySQL JDBC, например com.mysql.cj.jdbc.Driver.
userДа-Имя пользователя OceanBase. Включите сведения о тенанте в формате имени пользователя, требуемом OceanBase.
passwordДа-Пароль пользователя OceanBase.
databaseДа-Имя базы данных OceanBase.
include_tablesНет-Имена синхронизируемых таблиц, разделенные запятыми. Если параметр не указан, синхронизируются все таблицы базы данных.
exclude_tablesНет-Имена таблиц, которые не нужно синхронизировать, разделенные запятыми. Параметр действует только тогда, когда include_tables не задан.
table.<table_name>.target_tableНетИмя исходной таблицыЗадает имя целевой таблицы Doris для исходной таблицы.
table.<table_name>.exclude_columnsНет-Исключает неключевые столбцы исходной таблицы, разделенные запятыми. Каждый столбец должен существовать и не может быть столбцом первичного ключа.
offsetНетlatestНачальная позиция. initial: полная и инкрементальная синхронизация; snapshot: только полная синхронизация; earliest: начать с самой ранней доступной позиции binlog; latest: синхронизировать только изменения после запуска задания. Также поддерживается точная позиция, например {"file":"binlog.000001","pos":"154"}.
snapshot_split_sizeНет40960Количество строк в каждом split на этапе полной синхронизации. Должно быть положительным целым числом.
snapshot_parallelismНет1Максимальное число split, одновременно планируемых одним Task на этапе полной синхронизации. Должно быть положительным целым числом.
skip_snapshot_backfillНетtrueПропускать ли обратное заполнение binlog на этапе snapshot. Auto Table Creation Sync обеспечивает семантику At-least-once.
server_idНетСоздается автоматическиИдентификатор сервера CDC Reader. Поддерживается одиночное значение, например 5400, или замкнутый диапазон, например 5400-5408. Диапазон должен содержать не меньше идентификаторов, чем значение snapshot_parallelism.
ssl_modeНетdisableРежим SSL. Допустимые значения: disable, require и verify-ca.
ssl_rootcertТребуется условно-Требуется, если ssl_mode имеет значение verify-ca. Используйте формат FILE:<file_name> и сначала загрузите файл с помощью CREATE FILE.

Параметры schema, slot_name и publication_name не поддерживаются. Если указан любой из них, создание задания завершается ошибкой.

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

Оператор загрузки​

CREATE JOB <job_name>
[job_properties]
ON STREAMING
[COMMENT <comment>]
FROM OCEANBASE (
[source_properties]
)
TO DATABASE <target_db> (
[target_properties]
);
МодульОписание
job_nameИмя задания.
job_propertiesОбщие параметры задания, например max_interval.
commentКомментарий к заданию.
source_propertiesПараметры источника данных OceanBase.
target_propertiesПараметры целевой базы данных Doris.

Параметры целевой базы данных Doris​

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

Рекомендации и лучшие практики​

  • Перед расширением области синхронизации в рабочей среде используйте небольшой набор таблиц, чтобы проверить подключение к OBBinlog, сопоставление типов и задержку инкрементальной синхронизации.
  • Используйте include_tables для явного определения области синхронизации, чтобы добавленные позднее несвязанные таблицы не включались автоматически.
  • Исключайте неподдерживаемые неключевые столбцы с помощью table.<table_name>.exclude_columns. Столбцы первичного ключа исключать нельзя.
  • Перед внесением неподдерживаемого изменения схемы приостановите задание, примените совместимое изменение к таблице Doris, а затем возобновите задание.
  • Контролируйте CurrentOffset, LagBytes и ErrorMsg, чтобы убедиться, что задание продолжает выполняться и не было автоматически приостановлено.

Часто задаваемые вопросы​

Q1: Поддерживается ли режим совместимости OceanBase с Oracle?

Нет. При создании задания Doris проверяет ob_compatibility_mode и отклоняет любой режим, кроме режима совместимости с MySQL.

Q2: Поддерживается ли SQL Mapping Sync?

Нет. В настоящее время OceanBase поддерживает только Auto Table Creation Sync с использованием FROM OCEANBASE (...) TO DATABASE (...).

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

Установите для offset значение latest. Строки, существующие до перехода задания в состояние RUNNING, не загружаются через snapshot.

Q4: Перезаписывается ли уже существующая целевая таблица?

Нет. Автоматическое создание таблиц пропускает уже существующую целевую таблицу. Убедитесь, что ее первичный ключ и типы столбцов совместимы с исходной таблицей.