データ統合は、MySQL や PolarDB などのデータベースから OSS へのすべてのデータのリアルタイム同期をサポートしています。このトピックでは、データ統合を使用して MySQL から OSS のデータレイクにデータをリアルタイムで同期する方法について説明します。
前提条件
サーバーレスリソースグループまたはデータ統合専用リソースグループが購入されていること。
MySQL データソースと ApsaraDB for OceanBase データソースが作成されていること。詳細については、「データ統合のデータソースを作成する」をご参照ください。
説明MySQL データソースでバイナリログ機能を有効にする必要があります。詳細については、「MySQL 環境を準備する」をご参照ください。
リソースグループとデータソース間のネットワーク接続が確立されていること。詳細については、「ネットワーク接続ソリューション」をご参照ください。
制限事項
ソースのプライマリキー列の値は NULL または空文字列にすることはできません。そうしないと、関連する同期タスクの実行時にエラーが報告される可能性があります。
データベースから OSS のデータレイクへのすべてのデータのリアルタイム同期でサポートされる機能
MySQL データソースのデータ構造の同期、MySQL データソースのデータベース内のすべてまたは特定のテーブルの既存データの 1 回限りの同期、および MySQL データソースの増分データの OSS のデータレイクへのリアルタイム同期をサポートします。
テーブル作成および列追加 DDL 操作によって生成されたデータ変更の同期をサポートし、データ同期中にソーステーブルの数とソーステーブルの列数に基づいて、ターゲットテーブルとターゲットテーブルの列の自動調整をサポートします。
重要テーブルの削除やテーブル名の変更などの DDL 操作によって生成されたデータ変更は同期できません。これらの DDL 操作がソーステーブルで実行されると、関連する同期タスクは失敗します。
ワークスペースと同じリージョンで Data Lake Formation (DLF) がアクティブ化されている場合、DLF でのメタデータベースとメタテーブルの自動作成をサポートします。データが OSS のデータレイクに保存されると、システムは自動的にメタデータベースとメタテーブルを作成します。
説明リージョン間のメタデータベース作成はサポートされていません。
手順
ステップ 1: 同期タスクの種類を選択する
データ統合ページに移動します。
DataWorks コンソールにログインします。上部のナビゲーションバーで、目的のリージョンを選択します。左側のナビゲーションウィンドウで、 を選択します。表示されたページで、ドロップダウンリストから目的のワークスペースを選択し、[データ統合に移動] をクリックします。
左側のナビゲーションウィンドウで [同期タスク] をクリックし、ページの上部にある [同期タスクの作成] をクリックして、同期タスクの作成ページに移動します。次の基本情報を構成します:
ソースと宛先:
MySQL→OSSタスク名: 同期タスクのカスタム名を指定します。
同期タイプ:
リアルタイムデータベース同期。同期ステップ: [完全同期] と [増分同期] を選択します。
ステップ 2: ネットワークとリソースを構成する
[ネットワークとリソースの構成] セクションで、同期タスクの [リソースグループ] を選択します。[タスクリソースの使用状況] に CU 数を割り当てることができます。
[ソースデータソース] で、追加した
MySQLデータソースを選択します。[ターゲットデータソース] で、追加したOSSデータソースを選択し、[接続性のテスト] をクリックします。
ソースとターゲットの両方のデータソースが正常に接続されたことを確認したら、[次へ] をクリックします。
ステップ 3: ターゲットの基本設定を構成する
書き込みフォーマット: [Hudi]、[Paimon]、[lceberg] の 3 つの書き込みフォーマットをサポートします。
ストレージパスの選択: 同期したデータを保存する OSS パスを選択します。
メタデータベースを作成する場所: システムが DLF にメタデータベースを作成できるようにするかどうかを選択できます。
説明メタデータベースは、ワークスペースと同じリージョンでアクティブ化された DLF でのみ自動的に作成できます。
ステップ 4: 同期するテーブルを選択する
このステップでは、[ソーステーブル] リストでデータを同期するテーブルを選択し、
アイコンをクリックして選択したテーブルを [選択したテーブル] リストに移動できます。
特定のテーブルを選択:
ソーステーブルの [データベースフィルター] と [テーブルフィルター] で、データベース名またはテーブル名の特徴的な情報を入力して、同期が必要なデータベースとテーブルをフィルターできます。同期が必要なすべてのデータベースとテーブルを選択し、
アイコンをクリックして [選択したテーブル] セクションに移動します。[選択したテーブル] セクションの [データベースフィルター] と [テーブルフィルター] で、データベース名またはテーブル名の特徴的な情報を入力して、同期が不要なデータベースとテーブルをフィルターできます。同期が不要なすべてのデータベースとテーブルを選択し、
アイコンをクリックして [ソーステーブル] セクションに戻します。
正規表現を使用してテーブルを選択 (実行時に正規表現によるテーブルの追加/削除をサポート):
[データベースフィルター] と [テーブルフィルター] で構成された正規表現を通じてテーブル情報をフィルターします。[選択の確認] をクリックして、同期する必要があるデータベースとテーブルデータを選択します。
説明たとえば、データベース名のプレフィックスが
aで、テーブル名のプレフィックスがorderであるデータベースとテーブル情報をフィルターする必要がある場合、[データベースフィルター] にa.*を、[テーブルフィルタリング] フィールドに、そしてテーブルフィルターフィールドにorder.*を入力できます。
ステップ 5: ターゲットテーブルのマッピングを構成する
データを同期するテーブルを選択すると、選択したテーブルが [ターゲットテーブルのマッピングルール] セクションに自動的に表示されます。ターゲットテーブルのプロパティはマッピングを待機しています。ソーステーブルとターゲットテーブル間のマッピングを手動で定義して、データの読み取りと書き込みの関係を決定する必要があります。次に、[アクション] 列の [リフレッシュ] をクリックできます。ソーステーブルとターゲットテーブル間のマッピングを直接リフレッシュできます。また、ターゲットテーブルに関連する設定を構成した後に、ソーステーブルとターゲットテーブル間のマッピングをリフレッシュすることもできます。
同期するテーブルを選択し、[バッチリフレッシュマッピング] をクリックできます。マッピングルールが構成されていない場合、デフォルトのテーブル名ルールは
${tableName}です。ターゲットに同じ名前のテーブルが存在しない場合、新しいテーブルが自動的に作成されます。[構成] ボタンを [カスタムターゲットデータベース名マッピング] 列でクリックして、ターゲットデータベース名ルールをカスタマイズできます。
組み込み変数と手動で入力した文字列を使用して、最終的なターゲットデータベース名を形成できます。組み込み変数を編集できます。たとえば、新しいデータベース名ルールを作成して、ソースデータベース名にサフィックスを追加してターゲットデータベース名にすることができます。
[ターゲットテーブル名のマッピングルールをカスタマイズ] 列の [編集] をクリックして、ビジネス要件に基づいてターゲットテーブル名のマッピングルールを構成できます。
組み込み変数と特定の文字列を連結して、ターゲットテーブル名にすることができます。組み込み変数を編集できます。たとえば、マッピングルールを作成するときに、ソーステーブル名を示す変数にサフィックスを追加して、ターゲットテーブル名を形成できます。
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] をクリックし、同期タスクの詳細な実行プロセスを表示します。
同期タスクの操作とメンテナンス
タスクの実行ステータスを表示する
同期タスクを作成した後、同期タスクページで現在作成されている同期タスクのリストと各同期タスクの基本情報を表示できます。
[アクション] 列で同期タスクを [開始] または [停止] したり、[その他] メニューから同期タスクを [編集]、[表示] したり、その他の操作を実行したりできます。
開始されたタスクについては、[実行概要] で基本的な実行ステータスを確認でき、対応する概要エリアをクリックして実行の詳細を表示することもできます。
MySQL から OSS へのリアルタイム同期タスクは、次の 3 つのステップで構成されます。
構造移行: ターゲットテーブルの作成方法 (既存のテーブル/自動テーブル作成) が含まれます。自動テーブル作成の場合は、DDL 文が表示されます。
完全初期化: オフライン同期テーブル情報、同期の進行状況、書き込まれたレコード数が含まれます。
リアルタイムデータ同期: リアルタイム同期統計情報 (リアルタイムの進行状況、DDL レコード、DML レコード、アラート情報) が含まれます。
同期タスクを再実行する
特別な場合、ソースにテーブルを追加または削除したり、ターゲットテーブルのスキーマや名前を変更したりした場合は、同期タスクの [アクション] 列の [その他] をクリックし、[再実行] をクリックして変更後にタスクを再実行できます。再実行プロセス中、同期タスクは新しく追加されたテーブルからターゲットにのみデータを同期するか、マッピングされたソーステーブルからスキーマまたは名前が変更されたターゲットテーブルにのみデータを同期します。
タスクの構成を変更せずに同期タスクを再実行する場合は、[アクション] 列の [その他] をクリックし、[再実行] をクリックしてタスクを再実行し、完全同期と増分同期を再度実行します。
タスクにテーブルを追加または削除した後に同期タスクを再実行する場合は、変更後に [完了] をクリックします。この場合、同期タスクの [アクション] 列に [更新の適用] が表示されます。[更新の適用] をクリックして、システムに同期タスクの再実行をトリガーします。再実行プロセス中、同期タスクは新しく追加されたテーブルからターゲットにデータを同期します。元のテーブルのデータは再度同期されません。