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

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

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

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

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

В настоящее время непрерывная загрузка OceanBase поддерживает только режим совместимости 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; см. Синхронизация Schema Change
Типы данныхИспользуется преобразование типов, совместимое с 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Число строк в каждом фрагменте на этапе полной синхронизации. Должно быть положительным целым числом.
snapshot_parallelismНет1Максимальное число фрагментов, планируемых параллельно одним Task на этапе полной синхронизации. Должно быть положительным целым числом.
skip_snapshot_backfillНетtrueПропускать ли обратное заполнение binlog на этапе создания снимка. Auto Table Creation Sync использует семантику at-least-once.
server_idНетСоздаётся автоматическиServer ID средства чтения CDC. Поддерживается одно значение, например 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Общие параметры Job, такие как max_interval и compute_group.
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, не загружаются через снимок.

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

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

Report issue