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

MySQL

Doris предоставляет следующие способы загрузки данных из MySQL:

  • Загрузка данных MySQL через JDBC Catalog

С помощью JDBC Catalog Doris представляет MySQL как внешний Catalog, позволяя выполнять SQL-запросы непосредственно к данным MySQL. В сочетании с INSERT INTO или CREATE TABLE AS SELECT этот способ подходит для разовой миграции или периодической пакетной загрузки.

  • Непрерывная синхронизация данных MySQL через Streaming Job

Doris использует Streaming Job для непрерывной полной и инкрементальной синхронизации данных из MySQL в Doris. Благодаря встроенным возможностям чтения Flink CDC Doris поддерживает непрерывную работу задания, читает Binlog из MySQL и записывает данные в таблицы Doris с семантикой exactly-once. Поддерживаются два режима: синхронизация с сопоставлением SQL и синхронизация с автоматическим созданием таблиц. Доступно начиная с Doris 4.1.0.

  • Загрузка данных MySQL через Flink Doris Connector

Используйте Flink Doris Connector совместно с Flink MySQL CDC для синхронизации в реальном времени. Этот способ подходит для сценариев, требующих дополнительной логики потоковой обработки во Flink. Коннектор также предоставляет инструмент для синхронизации всей базы данных одним действием. Подробнее см. в разделе Flink Doris Connector.

  • Загрузка данных MySQL с помощью сторонних инструментов

Инструменты интеграции данных, такие как DataX, SeaTunnel и CloudCanal, также поддерживают синхронизацию данных из MySQL в Doris.

В большинстве случаев для разовой миграции данных можно напрямую использовать JDBC Catalog. Если требуется непрерывная полная и инкрементальная синхронизация, рекомендуется Streaming Job.

Загрузка данных MySQL через JDBC Catalog​

Представьте MySQL как внешний Catalog с помощью JDBC Catalog, затем используйте INSERT INTO или CREATE TABLE AS SELECT для загрузки данных. Подробное описание синтаксиса см. в разделе JDBC MySQL Catalog.

Шаг 1: Подготовка данных в MySQL​

CREATE TABLE test.students (
    id     INT PRIMARY KEY,
    name   VARCHAR(64),
    age    INT
);

INSERT INTO test.students VALUES (1, 'Emily', 25), (2, 'Bob', 30);

Шаг 2: Создание Catalog в Doris​

CREATE CATALOG mysql_catalog PROPERTIES (
    "type"         = "jdbc",
    "user"         = "root",
    "password"     = "123456",
    "jdbc_url"     = "jdbc:mysql://127.0.0.1:3306/test",
    "driver_url"   = "mysql-connector-java-8.0.25.jar",
    "driver_class" = "com.mysql.cj.jdbc.Driver"
);

Шаг 3: Создание целевой таблицы в Doris​

CREATE DATABASE IF NOT EXISTS doris_db;

CREATE TABLE doris_db.students (
    id     INT,
    name   VARCHAR(64),
    age    INT
)
UNIQUE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES ("replication_num" = "1");

Шаг 4: Загрузка данных с помощью INSERT INTO​

INSERT INTO doris_db.students
SELECT id, name, age FROM mysql_catalog.test.students;

Если целевая таблица ещё не существует, можно также использовать CREATE TABLE AS SELECT, чтобы создать таблицу и загрузить данные за один шаг:

CREATE TABLE doris_db.students
PROPERTIES ("replication_num" = "1")
AS
SELECT * FROM mysql_catalog.test.students;

Шаг 5: Проверка загруженных данных​

SELECT * FROM doris_db.students;
+----+-------+------+
| id | name  | age  |
+----+-------+------+
|  1 | Emily |   25 |
|  2 | Bob   |   30 |
+----+-------+------+

Непрерывная синхронизация данных MySQL через Streaming Job​

Streaming Job непрерывно читает Binlog MySQL через Flink CDC и записывает данные в Doris. Поддерживаются два режима:

  • MySQL CDC с автоматическим созданием таблиц: Doris автоматически создаёт целевые таблицы при первой синхронизации (используйте include_tables для синхронизации одной, нескольких или всех таблиц). Обеспечивает семантику at-least-once.
  • MySQL CDC с сопоставлением SQL: целевая таблица должна быть заранее создана в Doris. Поддерживает гибкое сопоставление столбцов, преобразование данных и семантику exactly-once.

Ограничения​

  1. Поддерживаются только таблицы с первичным ключом (Unique Key).
  2. Требуется право Load. Для синхронизации с автоматическим созданием таблиц также требуется право Create, чтобы автоматически создавать целевые таблицы при первом запуске.
  3. Доступно начиная с Doris 4.1.0.

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

Перед отправкой задания Streaming Job необходимо включить Binlog на стороне MySQL и предоставить пользователю соответствующие права REPLICATION. Инструкции по настройке для конкретных сред см. в следующих разделах:

Пример работы: синхронизация с автоматическим созданием таблиц​

Синхронизация с автоматическим созданием таблиц использует синтаксис FROM MYSQL ... TO DATABASE .... Целью является база данных Doris, а целевые таблицы создаются автоматически при первой синхронизации.

Шаг 1: Подготовка данных в MySQL​

CREATE TABLE test.students (
    id     INT PRIMARY KEY,
    name   VARCHAR(64),
    age    INT
);

INSERT INTO test.students VALUES (1, 'Emily', 25), (2, 'Bob', 30);

Шаг 2: Создание целевой базы данных в Doris​

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

CREATE DATABASE IF NOT EXISTS doris_db;

Шаг 3: Создание задания Streaming Job​

В примере ниже параметр include_tables используется для синхронизации только таблицы students (несколько таблиц можно перечислить через запятую; оставьте значение пустым, чтобы синхронизировать всю базу данных):

CREATE JOB mysql_db_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" = "students",
    "offset"         = "initial"
)
TO DATABASE doris_db (
    "table.create.properties.replication_num" = "1"  -- set to 1 in single-BE deployments
);

Шаг 4: Проверка состояния задания​

SELECT * FROM jobs("type"="insert") WHERE ExecuteType = "STREAMING";

Шаг 5: Просмотр автоматически созданных таблиц Doris и загруженных данных​

SHOW TABLES FROM doris_db;
SELECT * FROM doris_db.students;

Другие распространённые операции и полный справочник параметров см. в разделе MySQL CDC с автоматическим созданием таблиц.

Пример работы: синхронизация с сопоставлением SQL​

Шаг 1: Подготовка данных в MySQL​

CREATE TABLE test.students (
    id     INT PRIMARY KEY,
    name   VARCHAR(64),
    age    INT
);

INSERT INTO test.students VALUES (1, 'Emily', 25), (2, 'Bob', 30);

Шаг 2: Создание целевой таблицы в Doris​

Синхронизация с сопоставлением SQL требует, чтобы целевая таблица была создана заранее:

CREATE DATABASE IF NOT EXISTS doris_db;

CREATE TABLE doris_db.students (
    id     INT,
    name   VARCHAR(64),
    age    INT
)
UNIQUE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES ("replication_num" = "1");

Шаг 3: Создание задания Streaming Job​

Используйте CREATE STREAMING JOB с синтаксисом INSERT INTO ... SELECT * FROM cdc_stream(...):

CREATE JOB mysql_students_sync
ON STREAMING
DO
INSERT INTO doris_db.students
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"     = "test",
    "table"        = "students",
    "offset"       = "initial"
);

Шаг 4: Проверка состояния задания​

SELECT * FROM jobs("type"="insert") WHERE ExecuteType = "STREAMING";

Шаг 5: Проверка загруженных данных​

SELECT * FROM doris_db.students;

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

Report issue