他の TP システムからのデータ移行

このドキュメントでは、MySQL、SQL Server、Oracle、PostgreSQL などの従来型トランザクション（OLTP / TP）データベースから 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、または MySQL 互換ソースから Doris への同期に対応する NineData を使用します。

---

オプション 1：Multi-Catalog（一度限りの移行に推奨）

JDBC Catalog を使用して TP データベースのテーブルを Doris の外部テーブルとしてマッピングし、INSERT INTO または CREATE TABLE AS SELECT (CTAS) を使用してデータロードを行います。

手順

1. ソース TP データベースを指す JDBC Catalog を作成します。
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 データインポートを参照してください。

---

オプション 2：Flink Doris Connector

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を参照してください。

2.2 リアルタイム同期：Flink CDC

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 をバッチ処理に使用しているシナリオに適しています。Spark Connector の JDBC Source で 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()

関連ドキュメント：

• JDBC とその他のデータベース
• Spark Doris Connector バッチ書き込み

---

オプション 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：サードパーティ同期ツール

上記のネイティブオプションに加えて、コミュニティで一般的に使用されているサードパーティ同期ツールも利用できます。既存の同期プラットフォームがある場合や、ビジュアルインターフェースでタスクを管理したい場合に適しています。

• NineData
• DataX
• SeaTunnel
• CloudCanal

---

FAQ

Q1: MySQL の数個のテーブルを一度だけ Doris に移行する場合、どのオプションを選ぶべきですか？

Multi-Catalog を使用します。JDBC Catalog を作成した後、INSERT INTO または CTAS を実行すれば、追加コンポーネントなしで移行を完了できます。

Q2: MySQL から Doris に増分データ（UPDATE / DELETE を含む）をリアルタイムで同期するにはどうすればよいですか？

Flink Doris Connector + Flink CDC を使用し、Doris Sink で sink.enable-delete = true を有効にするか、Doris に組み込まれた Streaming Job を使用して継続同期を行います。

Q3: MySQL データベース全体を一度の手順で Doris に同期するにはどうすればよいですか？

Flink Doris Connector の mysql-sync-database データベース全体同期機能、または Streaming Job の複数テーブル同期機能を使用します。いずれもスキーマとデータを一度の手順で同期できます。

Q4: Oracle、SQL Server、PostgreSQL もソースデータベースとしてサポートされていますか？

はい。Multi-Catalog と Flink / Spark Connector は、それぞれの JDBC ドライバを介して Oracle、SQL Server、PostgreSQL などの TP データベースをサポートします。Streaming Job は現在、MySQL と PostgreSQL をサポートしています。

Q5: 移行中に Doris テーブルのレプリケーション数を指定するにはどうすればよいですか？

CTAS ステートメントで PROPERTIES('replication_num' = 'N') を指定するか、Flink のデータベース全体同期コマンドで --table-conf replication_num=N を渡します。