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

Миграция данных из других TP-систем

В этом документе описывается миграция данных из традиционных транзакционных (OLTP / TP) баз данных, таких как MySQL, SQL Server, Oracle и PostgreSQL, в Apache Doris. Doris предоставляет несколько вариантов загрузки данных для одноразовой миграции, офлайн-пакетной синхронизации и синхронизации инкрементальных изменений в реальном времени.

Обзор вариантов миграции​

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

ВариантСценарий примененияРежим синхронизацииПоддержка всей базы данныхПоддержка инкремента / CDC
Multi-CatalogОдноразовая миграция, разовые запросы, создание таблицы CTASОфлайн / по запросуРеализуется с помощью SQLНет
Flink Doris Connector (JDBC)Офлайн-пакетная синхронизация в стеке FlinkОфлайнНетНет
Flink Doris Connector (CDC)Полная и инкрементальная синхронизация в реальном времениВ реальном времениДаДа (INSERT / UPDATE / DELETE)
Spark ConnectorПакетная запись данных в стеке SparkОфлайнНетНет
Streaming JobВстроенная в Doris непрерывная синхронизация из MySQL / PostgreSQLПотоковый (непрерывный)ДаДа
Сторонние инструменты (NineData / DataX / SeaTunnel / CloudCanal)Существующие платформы синхронизации, визуальное управлениеОфлайн / в реальном времениДаДа

Примечание: если нужно выполнить миграцию без подключения внешнего вычислительного движка, используйте Multi-Catalog или Streaming Job. Для синхронизации инкрементальных изменений в реальном времени используйте Flink Doris Connector + CDC, Streaming Job или NineData для совместимых с MySQL источников в Doris.

С помощью JDBC Catalog сопоставьте таблицы в TP-базе данных с внешними таблицами в Doris, затем загрузите данные с помощью INSERT INTO или CREATE TABLE AS SELECT (CTAS).

Шаги​

  1. Создайте JDBC Catalog, указывающий на исходную TP-базу данных.
  2. Используйте INSERT INTO для записи в существующую таблицу Doris или CTAS, чтобы за один шаг создать таблицу и загрузить данные.
  3. Проверьте количество строк и соответствие полей.

Пример: миграция данных из MySQL​

CREATE CATALOG mysql_catalog properties(
    'type' = 'jdbc',
    'user' = 'root',
    'password' = '123456',
    'jdbc_url' = 'jdbc:mysql://host:3306/mysql_db',
    'driver_url' = 'mysql-connector-java-8.0.25.jar',
    'driver_class' = 'com.mysql.cj.jdbc.Driver'
);

-- Load into an existing table with INSERT INTO
INSERT INTO internal.doris_db.tbl1
SELECT * FROM mysql_catalog.mysql_db.table1;

-- Create the table and load data in one step with CTAS
CREATE TABLE internal.doris_db.tbl1
PROPERTIES('replication_num' = '1')
AS
SELECT * FROM mysql_catalog.mysql_db.table1;

Дополнительные сведения см. в разделе Импорт данных Catalog.

С помощью Flink можно выполнять синхронизацию TP-систем в офлайн-пакетном режиме или режиме инкрементальной синхронизации в реальном времени.

2.1 Офлайн-синхронизация: Flink JDBC Source + Doris Sink​

Этот вариант подходит для офлайн-переноса данных в стеке Flink. Следующий пример Flink SQL показывает соответствующий шаблон:

CREATE TABLE student_source (
    id INT,
    name STRING,
    age INT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'jdbc',
  'url' = 'jdbc:mysql://localhost:3306/mydatabase',
  'table-name' = 'students',
  'username' = 'username',
  'password' = 'password'
);

CREATE TABLE student_sink (
    id INT,
    name STRING,
    age INT
    )
    WITH (
      'connector' = 'doris',
      'fenodes' = '127.0.0.1:8030',
      'table.identifier' = 'test.students',
      'username' = 'root',
      'password' = 'password',
      'sink.label-prefix' = 'doris_label'
);

INSERT into student_sink select * from student_source;

Дополнительные сведения см. в разделе Flink JDBC.

С помощью Flink CDC можно в одном задании выполнить полное чтение данных и захват инкрементальных изменений, а также синхронизировать события INSERT / UPDATE / DELETE.

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

SET 'execution.checkpointing.interval' = '10s';

CREATE TABLE cdc_mysql_source (
  id int
  ,name VARCHAR
  ,PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = '127.0.0.1',
'port' = '3306',
'username' = 'root',
'password' = 'password',
'database-name' = 'database',
'table-name' = 'table'
);

-- Supports synchronization of INSERT / UPDATE / DELETE events
CREATE TABLE doris_sink (
id INT,
name STRING
)
WITH (
  'connector' = 'doris',
  'fenodes' = '127.0.0.1:8030',
  'table.identifier' = 'database.table',
  'username' = 'root',
  'password' = '',
  'sink.properties.format' = 'json',
  'sink.properties.read_json_by_line' = 'true',
  'sink.enable-delete' = 'true',  -- Synchronize delete events
  'sink.label-prefix' = 'doris_label'
);

insert into doris_sink select id,name from cdc_mysql_source;

Синхронизация всей базы данных / нескольких таблиц​

Для синхронизации всей базы данных или нескольких таблиц из TP-базы Flink Doris Connector предоставляет готовую функцию синхронизации всей базы данных, которая за один шаг синхронизирует схему и записывает данные:

<FLINK_HOME>bin/flink run \
    -Dexecution.checkpointing.interval=10s \
    -Dparallelism.default=1 \
    -c org.apache.doris.flink.tools.cdc.CdcTools \
    lib/flink-doris-connector-1.16-24.0.1.jar \
    mysql-sync-database \
    --database test_db \
    --mysql-conf hostname=127.0.0.1 \
    --mysql-conf port=3306 \
    --mysql-conf username=root \
    --mysql-conf password=123456 \
    --mysql-conf database-name=mysql_db \
    --including-tables "tbl1|test.*" \
    --sink-conf fenodes=127.0.0.1:8030 \
    --sink-conf username=root \
    --sink-conf password=123456 \
    --sink-conf jdbc-url=jdbc:mysql://127.0.0.1:9030 \
    --sink-conf sink.label-prefix=label \
    --table-conf replication_num=1

Дополнительные сведения см. в разделе Синхронизация всей базы данных.

Вариант 3: Spark Connector​

Этот вариант подходит для сценариев, в которых Spark уже используется для пакетной обработки. С помощью JDBC Source в Spark Connector можно читать данные из TP-базы, а с помощью Doris Sink — записывать их в Doris.

val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:postgresql:dbserver")
  .option("dbtable", "schema.tablename")
  .option("user", "username")
  .option("password", "password")
  .load()

 jdbcDF.write.format("doris")
  .option("doris.table.identifier", "db.table")
  .option("doris.fenodes", "127.0.0.1:8030")
  .option("user", "root")
  .option("password", "")
  .save()

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

Вариант 4: Streaming Job (встроенная в Doris непрерывная синхронизация)​

В Doris предусмотрена встроенная возможность Streaming Job, которая непрерывно синхронизирует данные из MySQL или PostgreSQL без внешнего вычислительного движка.

Пример синхронизации нескольких таблиц​

CREATE JOB multi_table_sync
ON STREAMING
FROM MYSQL (
        "jdbc_url" = "jdbc:mysql://127.0.0.1:3306",
        "driver_url" = "mysql-connector-j-8.0.31.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"
)

Дополнительные сведения см. в разделе Непрерывная загрузка Postgres / MySQL.

Вариант 5: сторонние инструменты синхронизации​

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

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

Q1: Какой вариант выбрать для одноразовой миграции нескольких таблиц MySQL в Doris?

Используйте Multi-Catalog. После создания JDBC Catalog выполните INSERT INTO или CTAS, чтобы завершить миграцию без дополнительных компонентов.

Q2: Как в реальном времени синхронизировать инкрементальные данные (включая UPDATE / DELETE) из MySQL в Doris?

Используйте Flink Doris Connector + Flink CDC и включите sink.enable-delete = true в Doris Sink либо используйте встроенный Streaming Job в Doris для непрерывной синхронизации.

Q3: Как за один шаг синхронизировать всю базу данных MySQL с Doris?

Используйте функцию синхронизации всей базы данных mysql-sync-database в Flink Doris Connector или возможность синхронизации нескольких таблиц в Streaming Job. Оба варианта за один шаг синхронизируют схему и данные.

Q4: Поддерживаются ли в качестве исходных баз данных Oracle, SQL Server и PostgreSQL?

Да. Multi-Catalog и коннекторы Flink / Spark поддерживают TP-базы данных, такие как Oracle, SQL Server и PostgreSQL, через соответствующие JDBC-драйверы. Streaming Job в настоящее время поддерживает MySQL и PostgreSQL.

Q5: Как указать количество реплик таблицы Doris во время миграции?

Укажите PROPERTIES('replication_num' = 'N') в операторе CTAS или передайте --table-conf replication_num=N в команде синхронизации всей базы данных Flink.

Report issue