DTS、Flink CDC、カタログ、または DataWorks を使用して、MySQL ソース (自己管理型 MySQL、ApsaraDB RDS for MySQL、PolarDB for MySQL など) から ApsaraDB for SelectDB にデータを移行できます。データ量とユースケースに基づいて方法を選択してください。
移行方法の比較
DTS、DataWorks、Flink CDC、およびカタログはすべて MySQL データを ApsaraDB for SelectDB に移行できますが、それぞれの機能は異なります。ご使用のシナリオに最も適した方法を選択してください。
方法 | 履歴データ移行 | 増分データ同期 | スキーマ移行 | データベース移行 | DDL 同期 | データ検証 |
DTS | ✔️ | ✔️ | ✔️ | ✔️ | ✔️ | ✔️ |
DataWorks | ✔️ | ✔️ | ✔️ | ✔️ | ✔️ | ❌ |
Flink CDC | ✔️ | ✔️ | ✔️ | ✔️ | ✔️ | ❌ |
カタログ | ✔️ | ❌ | ❌ | ❌ | ❌ | ❌ |
以下のセクションでは、各方法の基本的な手順を説明します。詳細については、各方法のドキュメントをご参照ください。
前提条件
MySQL インスタンスと SelectDB インスタンスがネットワーク経由で通信できることを確認してください。
MySQL インスタンスと SelectDB インスタンスは同じ VPC 内にあります。そうでない場合は、まずネットワーク接続の問題を解決してください。詳細については、「ApsaraDB for SelectDB インスタンスとデータソース間のネットワーク接続に関する問題の解決方法」をご参照ください。
MySQL インスタンスの IP アドレスを、SelectDB の IP アドレスホワイトリストに追加します。詳細については、「IP アドレスホワイトリストを設定する」をご参照ください。
MySQL インスタンスに IP アドレスホワイトリストがある場合は、SelectDB インスタンスの VPC の IP CIDR ブロックを MySQL の IP アドレスホワイトリストに追加します。
-
ご利用の SelectDB インスタンスが存在する VPC の IP 範囲を取得するには、「ApsaraDB for SelectDB インスタンスが配置されている VPC の IP CIDR ブロックを確認する方法」をご参照ください。
-
SelectDB インスタンスのパブリック IP アドレスを取得するには、そのパブリックエンドポイントSelectDB に対して
PINGコマンドを使用します。
-
DTS を使用したデータ移行
DTS は、MySQL から ApsaraDB for SelectDB への履歴データ移行と増分データ同期をサポートし、データベースとテーブルの移行、DDL 同期、データ検証などの機能を提供します。 次の例では、ApsaraDB RDS for MySQL インスタンスを使用します。 詳細については、「DTS を使用してデータをインポートする」をご参照ください。
操作手順
ApsaraDB for SelectDB コンソールにログインします。
画面左上で、インスタンスが存在するリージョンを選択します。
インスタンスリスト ページで、対象の インスタンス ID をクリックし、インスタンスの詳細 ページに移動します。
左側のナビゲーションペインで、データの移行と同期 をクリックし、Data Synchronization タブを選択します。
説明DTS データ同期は、1 回限りの履歴データ移行と、リアルタイムの増分データ同期を組み合わせたものです。
[同期タスクの作成] をクリックし、ソースデータベースと宛先データベースを設定します。
設定を完了したら、ページ下部にある[接続をテストして続行]をクリックします。
タスクオブジェクトおよび詳細設定を構成します。
データベースレベルでオブジェクトを選択した場合、DTS はタスクの開始後に作成されたテーブルのデータを同期しません。同期中にテーブルを追加する予定がある場合は、テーブルレベルでオブジェクトを選択してください。後で [同期オブジェクトの変更] を使用して選択内容を変更できます。
テーブルレベルでオブジェクトを選択し、テーブル名や列名のマッピングなどの編集が必要な場合、1 つの同期タスクでサポートされるテーブルは最大 1,000 個です。この制限を超えると、タスクの送信時にエラーが報告されます。この場合、同期するテーブルをバッチに分割して複数のタスクを設定するか、データベースレベルの同期タスクを設定してください。
同期先のインスタンスに同期するオブジェクトの名前を変更するには、[選択済みオブジェクト] セクションでオブジェクトを右クリックします。 詳細については、「単一オブジェクトの名前のマッピング」をご参照ください。
一度に複数のオブジェクトの名前を変更するには、[選択済みオブジェクト] セクションの右上隅にある [一括編集] をクリックします。 詳細については、「複数オブジェクト名の一括マッピング」をご参照ください。
データベースまたはテーブルレベルで同期する SQL 操作を選択するには、選択中のオブジェクト ボックスで同期するオブジェクトを右クリックし、表示されるダイアログボックスで目的の SQL 操作を選択します。 サポートされている操作の一覧については、「増分同期をサポートする SQL 操作」をご参照ください。
WHERE 句を使用してデータをフィルターするには、選択中のオブジェクト ボックスで同期するテーブルを右クリックし、表示されるダイアログボックスでフィルター条件を設定します。詳細については、「フィルター条件を設定する」をご参照ください。
オブジェクト名マッピング機能を使用する場合、名前が変更されたオブジェクトに依存する他のオブジェクトの同期が失敗する可能性があります。
オプション: 上記の設定が完了したら、次：データベースおよびテーブルのフィールド設定 をクリックして、移行先テーブルの プライマリキー列の追加、配布キー、および エンジンの選択 を設定します。
このステップは、タスクオブジェクトを設定する際に、同期タイプ に スキーマ同期 を選択した場合にのみ利用可能です。定義ステータス を すべて に設定し、その後で設定を変更できます。
複数の列を選択して複合 プライマリキー列の追加 を構成できます。また、プライマリキー列の追加 から 1 つ以上の列を 配布キー として選択する必要があります。エンジンの選択 については、Unique のみがサポートされています。
タスクを保存して事前チェックを実行します。[事前チェック成功率] が 100% になったら、[次へ: 購入] をクリックします。
[購入] ページで、データ同期インスタンスの課金方法とリンク仕様を選択します。[データ伝送サービス (従量課金) サービス規約] を読み、選択します。[購入して開始] をクリックして同期タスクを開始します。データ同期ページでタスクの進捗を確認できます。
カテゴリ | 設定項目 | 説明 |
タスク情報 | [[タスク名]] | タスクの名前。 DTSはタスクに名前を自動的に割り当てます。 タスクを簡単に識別できるように、わかりやすい名前を指定することをお勧めします。 一意のタスク名を指定する必要はありません。 |
ソースデータベース | [データベースタイプ] | MySQL を選択します。 |
[アクセス方法] | Alibaba Cloud インスタンスを選択します。 | |
[[インスタンスのリージョン]] | ソースの ApsaraDB RDS for MySQL インスタンスが存在するリージョンを選択します。 | |
[[RDS インスタンス ID]] | ソース ApsaraDB RDS for MySQL インスタンスの ID を選択します。 例: rm-2z3m****。 | |
[[データベースアカウント]] | ソース ApsaraDB RDS for MySQL インスタンスのデータベースアカウントを入力します。 必要な権限については、「データベースアカウントに必要な権限」をご参照ください。 | |
[[データベースのパスワード]] | データベースアカウントのパスワードを設定します。 | |
[暗号化] | データベースへの接続を暗号化するかどうかを指定します。 ビジネス要件に基づいて、[非暗号化] または [SSL暗号化] を選択します。 SSL暗号化を選択した場合、データ同期タスクを設定する前に、ApsaraDB RDS for MySQLインスタンスのSSL暗号化を有効にする必要があります。 詳細については、「SSL暗号化機能の設定」をご参照ください。 | |
ターゲットデータベース | [データベースタイプ] | [SelectDB] を選択します。 |
[アクセス方法] | Alibaba Cloud インスタンス を選択します。 | |
[インスタンスのリージョン] | ターゲットの SelectDB インスタンスが存在するリージョンを選択します。 | |
[インスタンス ID] | ターゲットの SelectDB インスタンスの ID を選択します。 | |
[データベースアカウント] | 移行先の SelectDB インスタンスのデータベースアカウントを入力します。必要な権限については、「データベースアカウントに必要な権限」をご参照ください。 | |
[データベースのパスワード] | データベースアカウントのパスワードを設定します。 |
設定項目 | 説明 |
[同期タイプ] | デフォルトでは、増分データ同期が選択されています。 [スキーマ同期] および [完全データ同期] も選択する必要があります。 事前チェックが完了すると、DTSは選択したオブジェクトの履歴データをソースインスタンスからターゲットインスタンスに同期します。 履歴データは、その後の増分同期の基礎となる。 重要 MySQL から SelectDB にデータが同期されると、データ型が変換されます。スキーマ同期 を選択しない場合は、まず宛先の SelectDB インスタンスに、Unique Key モデルを使用し、対応するスキーマを持つテーブルを作成する必要があります。詳細については、「データ型のマッピング」および「データモデル」をご参照ください。 |
[ソースオブジェクト] | [ソースオブジェクト] ボックスで、同期するオブジェクトをクリックし、 重要 |
[選択中のオブジェクト] | 説明 |
をクリックしてオブジェクトを [選択済みオブジェクト] ボックスに移動します。オブジェクトは、データベース、テーブル、または列レベルで選択できます。DataWorks を使用したデータ移行
DataWorks データ統合は、MySQL から ApsaraDB for SelectDB へデータを同期できます。詳細については、「DataWorks を使用してデータをインポートする」をご参照ください。
BITMAP、HLL (HyperLogLog)、または QUANTILE_STATE データ型のフィールドへの書き込みはサポートされていません。
データソースの追加
データ同期タスクを開発する際、DataWorks で MySQL データソースと SelectDB データソースを作成します。
MySQL データソースを作成します。詳細については、「MySQL データソース」をご参照ください。
SelectDB データソースを作成します。詳細については、「データソース管理」をご参照ください。次の表では、SelectDB データソースの設定パラメーターの一部について説明します。
パラメーター
説明
[データソース名]
データソースの名前です。
[MySQL接続URL]
JDBC 接続文字列
jdbc:mysql://<ip>:<port>/<dbname>を入力します。ApsaraDB for SelectDB コンソールの インスタンスの詳細 > ネットワーク情報 ページから、VPC アドレス (または インターネットアドレス) と MySQL プロトコルポート を取得できます。
例：
jdbc:mysql://selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:9030/test_db[HTTP接続URL]
HTTP プロトコルエンドポイントを
<ip>:<port>形式で入力します。ApsaraDB for SelectDB コンソールの インスタンスの詳細 > ネットワーク情報 ページで、VPC アドレス (または インターネットアドレス) と HTTP プロトコルポート を取得できます。
例：
selectdb-cn-4xl3jv1****.selectdbfe.rds.aliyuncs.com:8080[ユーザー名]
ApsaraDB for SelectDB インスタンスのユーザー名を入力します。
[パスワード]
ApsaraDB for SelectDB インスタンスで指定されたユーザーのパスワードを入力します。
データ同期の設定
データ移行タスクを設定するには、次のドキュメントを参照してください。
Flink CDC を使用したデータ移行
Flink は、MySQL から SelectDB にデータを移行するために、Flink SQL、Flink CDC、DataStream の 3 つのメソッドを提供します。Flink CDC は、履歴データの移行、増分データ同期、データベースとテーブルの移行、および DDL 同期をサポートします。次の例では、Flink CDC を使用して MySQL から ApsaraDB for SelectDB にデータを同期します。詳細については、「Flink を使用してデータをインポートする」をご参照ください。
操作手順
環境の準備
Flink 環境をセットアップします。この例では、Flink 1.16 スタンドアロン環境を使用します。
flink-1.16.3-bin-scala_2.12.tgz をダウンロードして解凍します。このバージョンが入手できない場合は、別のバージョンをダウンロードします。詳細については、「Apache Flink」をご参照ください。
wget https://archive.apache.org/dist/flink/flink-1.16.3/flink-1.16.3-bin-scala_2.12.tgz tar -zxvf flink-1.16.3-bin-scala_2.12.tgzFLINK_HOME/lib ディレクトリに移動して、flink-sql-connector-mysql-cdc-2.4.2 と flink-doris-connector-1.16-1.5.2 をダウンロードします。
cd flink-1.16.3 cd lib/ wget https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.4.2/flink-sql-connector-mysql-cdc-2.4.2.jar wget https://repo.maven.apache.org/maven2/org/apache/doris/flink-doris-connector-1.16/1.5.2/flink-doris-connector-1.16-1.5.2.jarFlink Standalone クラスターを起動します。
bin/start-cluster.shApsaraDB for SelectDB インスタンスを作成します。詳細については、「インスタンスの作成」をご参照ください。
MySQL プロトコル経由で ApsaraDB for SelectDB インスタンスに接続します。詳細については、「インスタンスに接続する」をご参照ください。
テストデータベースとテストテーブルを作成します。
テストデータベースを作成します。
CREATE DATABASE test_db;テストテーブルを作成します。
USE test_db; CREATE TABLE employees ( emp_no int NOT NULL, birth_date date, first_name varchar(20), last_name varchar(20), gender char(2), hire_date date ) UNIQUE KEY(`emp_no`) DISTRIBUTED BY HASH(`emp_no`) BUCKETS 1;
Flink CDC ジョブの送信
次の構文を使用して Flink CDC ジョブを送信します。
<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-1.5.2.jar \
mysql-sync-database \
--database test_db \
--including-tables "tbl1|test.*" \
--mysql-conf hostname=127.0.0.1 \
--mysql-conf username=root \
--mysql-conf password=123456 \
--mysql-conf database-name=mysql_db \
--sink-conf fenodes=selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080 \
--sink-conf username=admin \
--sink-conf password=****パラメーター | 説明 |
execution.checkpointing.interval | Flink のチェックポイント間隔です。データ同期の頻度に影響します。10s を推奨します。 |
parallelism.default | Flink ジョブの並列度です。並列度を増やすことで、データ同期のスループットを向上させることができます。 |
database | SelectDB のターゲットデータベースの名前です。 |
including-tables | 同期する MySQL テーブルです。複数のテーブルは縦棒 (|) を使って区切ることができ、正規表現もサポートされています。たとえば、 |
excluding-tables | 同期しないテーブルです。設定方法は including-tables パラメーターと同じです。 |
mysql-conf | MySQL CDC ソースの設定です。詳細については、「MySQL CDC コネクタ」をご参照ください。 |
sink-conf | Doris シンクのすべての設定。詳細については、「Flink を使用してデータをインポートする」をご参照ください。 |
table-conf | SelectDB テーブルの設定項目です。これらは CREATE TABLE ステートメントの PROPERTIES 句内の内容に対応します。 |
データを同期するには、flink-sql-connector-mysql-cdc-${version}.jar や flink-sql-connector-oracle-cdc-${version}.jar などの必要な Flink CDC 依存関係を $FLINK_HOME/lib ディレクトリに追加する必要があります。
データベースレベルの同期には、Flink 1.15 以降が必要です。Flink Doris Connector のダウンロードについては、「Flink Doris Connector」をご参照ください。
カタログを使用したデータ移行
SelectDB のカタログ機能は、フェデレーテッドクエリを使用して MySQL にアクセスし、履歴データの移行を簡素化します。次の例では、MySQL データベースから ApsaraDB for SelectDB にデータを同期します。詳細については、「JDBC データソース」をご参照ください。
操作手順
SelectDB インスタンスに接続します。詳細については、「インスタンスに接続する」をご参照ください。
説明DMS を使用して接続する場合、
SWITCHステートメントは機能しません。MySQL クライアントで接続することをお勧めします。MySQL JDBC カタログを作成します。
CREATE CATALOG jdbc_mysql PROPERTIES ( "type"="jdbc", "user"="root", "password"="123456", "jdbc_url" = "jdbc:mysql://127.0.0.1:3306/demo", "driver_url" = "mysql-connector-java-8.0.25.jar", "driver_class" = "com.mysql.cj.jdbc.Driver", "checksum" = "fdf55dcef04b09f2eaf42b75e61ccc9a" )パラメーターの説明
パラメーター
必須
デフォルト
説明
user
はい
なし
データベースアカウントです。
password
はい
なし
データベースアカウントのパスワードです。
jdbc_url
はい
なし
JDBC 接続文字列です。
driver_url
はい
なし
JDBC ドライバー JAR パッケージのファイル名です。
driver_class
はい
なし
JDBC ドライバーのクラス名です。
lower_case_table_names
いいえ
"false"
外部 JDBC データソースからデータベース名とテーブル名を小文字で同期するかどうかを指定します。
only_specified_database
いいえ
"false"
指定されたデータベースのみを同期するかどうかを指定します。
include_database_list
いいえ
""
only_specified_database=trueの場合、同期するデータベースのカンマ区切りリストを指定します。データベース名では大文字と小文字が区別されます。exclude_database_list
いいえ
""
only_specified_database=trueの場合、同期から除外するデータベースのリストをカンマ区切りで指定します。データベース名では大文字と小文字が区別されます。SelectDB に宛先テーブルを作成し、
INSERT INTO ... SELECTステートメントを使用してデータを同期します。INSERT INTOの詳細については、「INSERT INTO」をご参照ください。
# ターゲットテーブルを作成します。 CREATE TABLE selectdb_table ... # データを移行します。 INSERT INTO selectdb_table SELECT * FROM mysql_catalog.mysql_database.mysql_table;