このトピックでは、DataWorks のデータ統合を使用して、MySQL から OSS-HDFS のデータレイクにデータをリアルタイムで同期する方法について説明します。
前提条件
サーバーレスリソースグループ または データ統合専用リソースグループ が購入されていること。
MySQL データソースと ApsaraDB for OceanBase データソースが作成されていること。 詳細については、「データ統合のデータソースを作成する」をご参照ください。
説明MySQL データソースでバイナリログ機能を有効にする必要があります。 詳細については、「MySQL 環境の準備」をご参照ください。
リソースグループとデータソース間のネットワーク接続が確立されていること。 詳細については、「ネットワーク接続ソリューション」をご参照ください。
制限事項
ソースのプライマリキー列の値は NULL または空の文字列にできません。 そうしないと、関連する同期タスクの実行時にエラーが報告される可能性があります。
データベース全体からデータレイクへのリアルタイム同期の機能
この例で作成した同期タスクを使用して、MySQL データソースのデータ構造を OSS-HDFS のデータレイクに同期できます。 また、同期タスクを使用して、MySQL データベース内のすべてまたは特定のテーブルから Object Storage Service (OSS) のデータレイクに一度に既存データを同期し、MySQL データベースからデータレイクに増分データをリアルタイムで同期することもできます。
この例で作成された同期タスクは、ソースで実行された 列の追加 操作によって生成されたデータ変更を同期できます。 同期タスクの実行開始後にソースに新しいテーブルが追加された場合、システムはデータ同期中に宛先にテーブルを自動的に追加します。
重要テーブルの削除やテーブル名の変更などの DDL 操作によって生成されたデータ変更は同期できません。 これらの DDL 操作がソーステーブルで実行されると、関連する同期タスクは失敗します。
現在のアカウントを使用して Data Lake Formation (DLF) サービスをアクティベートしている場合、システムは DLF に関連するメタデータベースとメタデータテーブルを自動的に作成します。
手順
1. 同期タスクの種類を選択する
データ統合ページに移動します。
DataWorks コンソールにログインします。 上部のナビゲーションバーで、目的のリージョンを選択します。 左側のナビゲーションウィンドウで、 を選択します。 表示されたページで、ドロップダウンリストから目的のワークスペースを選択し、[データ統合に移動] をクリックします。
左側のナビゲーションウィンドウで、[同期タスク] をクリックします。 次に、ページの上部にある [同期タスクの作成] をクリックします。 表示されたページで、次の基本情報を設定します。
[ソースと宛先]:
MySQL→OSS-HDFS[新しいノード名]: 同期タスクの名前を指定します。
[同期方法]:
データベース全体のリアルタイム移行。[同期モード]: [完全初期化] と [増分同期] を選択します。
2. ネットワークとリソースを設定する
[ネットワークとリソースの設定] セクションで、同期タスクの [リソースグループ] を選択します。 タスクが [占有] できる計算ユニット (CU) の数を割り当てることができます。
[ソース] には追加した
MySQLデータソースを選択し、[宛先] には追加したOSS-HDFSデータソースを選択して、[接続性のテスト] をクリックします。
ソースと宛先のデータソースが接続されていることを確認したら、[次へ] をクリックします。
3. 宛先の基本設定を行う
[書き込みフォーマット]: [Hudi]、[Paimon]、[lceberg] のいずれかのフォーマットを選択できます。
[ストレージパスの選択]: 同期されたデータを格納する OSS 内のパスを選択します。
[メタデータベースを作成する場所]: DLF にメタデータベースを自動的に作成するかどうかを選択できます。
説明メタデータベースは、ワークスペースと同じリージョンでアクティベートされた DLF でのみ自動的に作成できます。
4. 同期するデータベースとテーブルを選択する
このステップでは、[ソーステーブル] リストで同期するテーブルを選択し、
アイコンをクリックして選択したテーブルを [選択したテーブル] リストに移動できます。
[特定のデータベースとテーブルを選択]:
[ソーステーブル] セクションの [データベースフィルター] および [テーブルフィルター] フィールドに、データベースまたはテーブル名の特徴を入力して、同期するデータベースとテーブルをフィルターできます。 同期するすべてのデータベースとテーブルを選択し、
アイコンをクリックして [選択したテーブル] セクションに移動します。[選択したテーブル] セクションで、[データベースフィルター] および [テーブルフィルター] フィールドにデータベースまたはテーブル名の特徴を入力して、同期したくないデータベースとテーブルをフィルターできます。 同期したくないすべてのデータベースとテーブルを選択し、
アイコンをクリックして [ソーステーブル] セクションに移動します。
[正規表現を使用してテーブルを選択 (実行時に正規表現を使用してテーブルを追加または削除することをサポート)]:
[データベースフィルター] および [テーブルフィルター] フィールドで設定した正規表現を使用してテーブルをフィルターできます。 [選択の確認] をクリックして、同期するデータベースとテーブルを選択します。
説明たとえば、名前が
aで始まるデータベースと、名前がorderで始まるテーブルをフィルターする場合、[データベースフィルター]a.*[テーブルフィルター] フィールドにorder.*を、テーブルフィルターフィールドに入力します。
5. ソーステーブルと宛先テーブルのマッピングを設定する
同期するテーブルを選択すると、選択したテーブルが [宛先テーブルのマッピングルール] セクションに自動的に表示されます。 宛先テーブルのプロパティはマッピングされるのを待っています。 データの読み取りと書き込みの関係を決定するには、ソーステーブルと宛先テーブルの間のマッピングを手動で定義する必要があります。 次に、[アクション] 列の [リフレッシュ] をクリックします。 ソーステーブルと宛先テーブル間のマッピングを直接リフレッシュできます。 宛先テーブルに関連する設定を行った後で、ソーステーブルと宛先テーブル間のマッピングをリフレッシュすることもできます。
同期するテーブルを選択し、[マッピング結果の一括リフレッシュ] をクリックします。 マッピングルールが設定されていない場合、デフォルトのテーブル名ルールは
${table name}です。 宛先に同じ名前のテーブルが存在しない場合、システムは自動的にテーブルを作成します。[宛先データベース名のマッピングルールを変更] 列で、[設定] をクリックして、宛先データベース名ルールをカスタマイズできます。
組み込み変数と指定された文字列を連結して、最終的な宛先データベース名にすることができます。 組み込み変数を編集できます。 たとえば、マッピングルールを作成するときに、ソースデータベース名を示す変数にサフィックスを追加して、宛先データベース名を形成できます。
[宛先テーブル名のマッピングルールをカスタマイズ] 列の [編集] をクリックして、ビジネス要件に基づいて宛先テーブル名のマッピングルールを設定できます。
組み込み変数と特定の文字列を連結して、宛先テーブル名にすることができます。 組み込み変数を編集できます。 たとえば、マッピングルールを作成するときに、ソーステーブル名を示す変数にサフィックスを追加して、宛先テーブル名を形成できます。
a. フィールドのデータ型マッピングを変更する
ソースフィールドのデータ型と宛先フィールドのデータ型の間には、デフォルトのマッピングが存在します。 [宛先テーブルのマッピングルール] セクションの右上隅にある [フィールドデータ型のマッピングを編集] をクリックして、ビジネス要件に基づいてソースフィールドと宛先フィールド間のデータ型マッピングを設定できます。 設定が完了したら、[適用してマッピングを更新] をクリックします。
b. 宛先テーブルのスキーマを変更してテーブルにフィールドを追加し、フィールドに値を割り当てる
宛先テーブルが [作成予定] 状態の場合、次の手順を実行してテーブルにフィールドを追加し、フィールドに値を割り当てることができます。
1 つ以上の宛先テーブルにフィールドを追加します。
単一の宛先テーブルにフィールドを追加する: フィールドを追加する宛先テーブルを見つけ、[宛先テーブル名] 列の
アイコンをクリックします。 表示されるダイアログボックスで、フィールドを追加します。一度に複数の宛先テーブルにフィールドを追加する: 一度にフィールドを追加する宛先テーブルを選択し、ページの下部にある [一括変更] をクリックし、次に [宛先テーブルスキーマ - フィールドの一括変更と追加] をクリックします。
フィールドに値を割り当てます。 次のいずれかの操作を実行して、フィールドに値を割り当てることができます。
単一の宛先テーブルに追加されたフィールドに値を割り当てる: 新しく追加されたフィールドに値を割り当てる宛先テーブルを見つけ、[値の割り当て] 列の [設定] をクリックします。 [追加フィールド] ダイアログボックスで、フィールドに値を割り当てます。
一度に複数の宛先テーブルに追加されたフィールドに値を割り当てる: 新しく追加されたフィールドに値を割り当てる宛先テーブルを選択し、ページの下部にある [一括変更] をクリックし、次に [値の割り当て] をクリックして、選択した宛先テーブルの同じフィールドに一度に値を割り当てます。
説明
アイコンをクリックして値の割り当て方法を切り替え、宛先テーブルに追加されたフィールドに定数と変数を割り当てることができます。
c. DML 処理ルールを設定する
データ統合は、デフォルトの DML 処理ルールを提供します。 ビジネス要件に基づいて、宛先テーブルの DML 処理ルールを設定することもできます。
単一の宛先テーブルの DML 処理ルールを設定する: DML 処理ルールを設定する宛先テーブルを見つけ、[DML ルールの設定] 列の [設定] をクリックして、テーブルの DML 処理ルールを設定します。
一度に複数の宛先テーブルの DML 処理ルールを設定する: DML 処理ルールを設定する宛先テーブルを選択し、ページの下部にある [一括変更] をクリックし、次に [DML ルールの設定] をクリックします。
6. アラートルールを設定する
同期タスクの失敗によるビジネスデータの同期の遅延を防ぐために、同期タスクにさまざまなアラートルールを設定できます。
ページの右上隅にある [アラートルールの設定] をクリックして [アラートルールの設定] パネルに移動します。
[アラートルールの設定] パネルで、[アラートルールの追加] をクリックします。 [アラートルールの追加] ダイアログボックスで、パラメーターを設定してアラートルールを設定します。
説明このステップで設定したアラートルールは、同期タスクによって生成されるリアルタイム同期サブタスクに対して有効になります。 同期タスクの設定が完了したら、リアルタイム同期タスクの管理 を参照して、リアルタイム同期タスクページに移動し、リアルタイム同期サブタスクに設定されたアラートルールを変更できます。
アラートルールを管理します。
作成されたアラートルールを有効または無効にできます。 アラートの重大度レベルに基づいて、異なるアラート受信者を指定することもできます。
7. 高度なパラメーターを設定する
ビジネス要件に基づいて、同期タスクに設定された特定のパラメーターの値を変更できます。 たとえば、[最大読み取り接続数] パラメーターに適切な値を指定して、現在の同期タスクがソースデータベースに過度の圧力をかけるのを防ぎ、データ生成が影響を受けないようにすることができます。
予期しないエラーやデータ品質の問題を防ぐために、パラメーターの値を変更する前に、パラメーターの意味を理解することをお勧めします。
設定ページの右上隅にある [高度なパラメーターの設定] をクリックします。
[高度なパラメーターの設定] パネルで、目的のパラメーターの値を変更します。
8. DDL 処理ルールを設定する
DDL 操作はソースで実行される場合があります。 ページの右上隅にある [DDL 機能の設定] をクリックして、ビジネス要件に基づいてソースからの DDL メッセージを処理するルールを設定できます。
詳細については、「DDL メッセージを処理するルールを設定する」をご参照ください。
9. リソースグループを設定する
ページの右上隅にある [リソースグループの設定] をクリックして、現在の同期タスクの実行に使用されるリソースグループを表示および変更できます。
10. 同期タスクを実行する
同期タスクの設定が完了したら、ページの下部にある [完了] をクリックします。
[データ統合] ページの [ノード] セクションで、作成した同期タスクを見つけ、[アクション] 列の [開始] をクリックします。
[タスク] セクションで同期タスクの [名前または ID] をクリックし、同期タスクの詳細な実行プロセスを表示します。
同期タスクの O&M 操作を実行する
同期タスクのステータスを表示する
同期タスクを作成した後、[同期タスク] ページで作成された同期タスクのリストと各同期タスクの基本情報を表示できます。
[アクション] 列の [開始] または [停止] をクリックして、同期タスクを開始または停止できます。 [アクション] 列の [詳細] をクリックし、[編集] または [表示] を選択して、同期タスクで他の操作を実行することもできます。
開始されたタスクについては、[実行概要] でタスクの基本的な実行ステータスを表示できます。 対応する概要エリアをクリックして、実行の詳細を表示することもできます。
MySQL データベース全体を OSS-HDFS に同期するためのリアルタイム同期タスクは、次の 3 つのステップで構成されます。
スキーマの移行: このタブには、宛先テーブルが新しく作成されたテーブルか既存のテーブルかなどの情報が表示されます。 新しく作成されたテーブルの場合、テーブルの作成に使用される DDL 文が表示されます。
完全なデータ初期化: このタブには、バッチ同期に関与するソーステーブルと宛先テーブル、同期の進行状況、同期されたデータレコードの数などの情報が表示されます。
リアルタイム同期: このタブには、同期の進行状況、DDL レコード、DML レコード、アラート情報など、リアルタイム同期に関する統計情報が表示されます。
同期タスクを再実行する
特殊なケースで、ソースにテーブルを追加または削除したり、宛先テーブルのスキーマまたは名前を変更したりした場合、同期タスクの [アクション] 列の [詳細] をクリックし、次に [再実行] をクリックして、変更後にタスクを再実行できます。 再実行プロセス中、同期タスクは、新しく追加されたテーブルから宛先へ、またはマップされたソーステーブルからスキーマまたは名前が変更された宛先テーブルへのみデータを同期します。
タスクの設定を変更せずに同期タスクを再実行する場合は、[アクション] 列の [詳細] をクリックし、次に [再実行] をクリックしてタスクを再実行し、完全同期と増分同期を再度実行します。
タスクにテーブルを追加または削除した後に同期タスクを再実行する場合は、変更後に [完了] をクリックします。 この場合、同期タスクの [アクション] 列に [更新の適用] が表示されます。 [更新の適用] をクリックして、システムに同期タスクの再実行をトリガーさせます。 再実行プロセス中、同期タスクは新しく追加されたテーブルから宛先にデータを同期します。 元のテーブルのデータは再度同期されません。