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

Миграция данных из других 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 либо используйте встроенный в Doris Streaming Job для непрерывной синхронизации.

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