Data Integration を使用すると、ApsaraDB for OceanBase、MySQL、Oracle、PolarDB、PolarDB-X 2.0 などのソースから MaxCompute にデータベース全体をリアルタイムで同期できます。このトピックでは、MySQL から MaxCompute へのデータ同期を例に、MySQL データベース全体から MaxCompute Delta テーブルへの完全データ同期と増分データ同期を実行する方法について説明します。
背景
MySQL データベース全体を MaxCompute にリアルタイムで同期する Data Integration タスクは、まずソース MySQL データベースから MaxCompute Delta テーブルに完全データを同期し、その後、増分データ変更をリアルタイムで同期します。宛先は、パーティションテーブルまたは非パーティションテーブルにすることができます。同期されたデータは、分レベルの待機時間で、ほぼリアルタイムで表示されます。Delta テーブルに同期された増分データは、最短 5 分でクエリできます。
MaxCompute Delta テーブルの詳細については、「準リアルタイムデータウェアハウス」をご参照ください。
前提条件
サーバーレスリソースグループまたはデータ統合専用リソースグループを購入済みであること。
MySQL データソースと MaxCompute データソースを作成済みであること。詳細については、「データソースの設定」をご参照ください。
MySQL: MySQL データソースでは、バイナリロギングを有効にする必要があります。詳細については、「MySQL データソースの準備」をご参照ください。
MaxCompute: MaxCompute データソースでは、同期タスクを実行する前に MaxCompute V2.0 データ型エディションを有効にする必要があります。MaxCompute V2.0 データ型エディションは、DECIMAL データ型をサポートしています。詳細については、「MaxCompute V2.0 データ型」をご参照ください。
リソースグループとデータソースは、ネットワーク経由で接続されている必要があります。詳細については、「ネットワーク接続ソリューションの概要」をご参照ください。
制限事項
この機能は、テナントレベルのスキーマ構文を使用する MaxCompute データソースをサポートしていません。
ソースデータから MaxCompute 外部テーブルへの同期はサポートされていません。
手順
ステップ 1: 同期タスクタイプを選択する
Data Integration ページに移動します。
DataWorks コンソールにログインします。上部のナビゲーションバーで、目的のリージョンを選択します。左側のナビゲーションウィンドウで、 を選択します。表示されたページで、ドロップダウンリストから目的のワークスペースを選択し、[Data Integration に移動] をクリックします。
左側のナビゲーションウィンドウで、[同期タスク] をクリックします。表示されたページで、[同期タスクの作成] をクリックし、次のパラメーターを設定します。
ソースと宛先:
MySQL→MaxComputeタスク名: 同期タスクのカスタム名を入力します。
同期タイプ:
データベース全体のリアルタイム同期。同期ステップ: [完全同期] と [増分同期] を選択します。
ステップ 2: ネットワークとリソースの設定を構成する
[ネットワークとリソースの設定] セクションで、同期タスクの [リソースグループ] を選択します。[タスクリソース使用量] セクションでタスクに CU を割り当てることもできます。
[ソースデータソース] で、
MySQLデータソースを選択します。[宛先データソース] で、MaxComputeデータソースを選択します。次に、[接続性のテスト] をクリックします。
両方のデータソースの接続性テストが成功したら、[次へ] をクリックします。
ステップ 3: 同期するデータベースとテーブルを選択する
[ソースデータベースとテーブル] エリアから同期するソーステーブルを選択し、
アイコンをクリックして右側の [選択されたデータベースとテーブル] エリアに移動します。
特定のテーブルを選択する:
[ソースデータベースとテーブル] エリアで、[データベースフィルター] および [テーブルフィルター] フィールドにキーワードを入力して、同期するデータベースとテーブルを検索します。データベースとテーブルを選択し、
アイコンをクリックして [選択されたデータベースとテーブル] エリアに移動します。[選択されたデータベースとテーブル] エリアで、[データベースフィルター] および [テーブルフィルター] フィールドにキーワードを入力して、同期したくないデータベースとテーブルを検索します。これらのデータベースとテーブルを選択し、
アイコンをクリックして [ソースデータベースとテーブル] エリアに戻します。
正規表現を使用してテーブルを選択する (タスク実行中のテーブルの追加または削除をサポート):
[データベースフィルター] および [テーブルフィルター] フィールドに正規表現を入力してテーブルをフィルターします。[選択の確認] をクリックして、同期するデータベースとテーブルを確認します。
説明たとえば、プレフィックス
aのデータベースとプレフィックスorderのテーブルでフィルターするには、[データベースフィルター] ボックスにa.*を入力し、[テーブルフィルター] ボックスにorder.*を入力します。
ステップ 4: 宛先テーブルをマッピングする
データを同期するテーブルを選択すると、選択したテーブルが [宛先テーブルのマッピングルール] セクションに自動的に表示されます。宛先テーブルのプロパティはマッピングされるのを待っています。ソーステーブルと宛先テーブル間のマッピングを手動で定義して、データの読み取りと書き込みの関係を決定する必要があります。次に、[アクション] 列の [リフレッシュ] をクリックします。ソーステーブルと宛先テーブル間のマッピングを直接リフレッシュできます。宛先テーブルに関連する設定を構成した後、ソーステーブルと宛先テーブル間のマッピングをリフレッシュすることもできます。
同期するテーブルを選択し、[マッピングの一括リフレッシュ] をクリックできます。マッピングルールが構成されていない場合、宛先テーブルのデフォルトの命名規則は
${source_database_name}_${table_name}です。生成された名前の宛先テーブルが存在しない場合、自動的に作成されます。[カスタムターゲットテーブル名マッピング] 列で、[編集] をクリックして、宛先テーブルの命名規則をカスタマイズできます。
組み込み変数とカスタム文字列を連結して、宛先テーブル名を作成できます。組み込み変数を編集することもできます。たとえば、ソーステーブル名にサフィックスを追加する命名規則を作成できます。
データを MaxCompute Delta テーブルに同期するには、宛先テーブルのプライマリキーを指定する必要があります。デフォルトでは、ソーステーブルのプライマリキーが使用されます。ソーステーブルにプライマリキーがない場合は、プライマリキー列をカスタマイズできます。プライマリキーを指定しないと、同期タスクを保存できません。
自動的に作成された Delta テーブルのデフォルトのバケット数は 16 です。この数は、宛先テーブルマッピングの [テーブルバケット数] 設定で変更できます。
既存のテーブルのバケット数は変更できません。バケット数は、テーブルデータがどのように分割されるかを決定します。クエリ、書き込み、データマージなどの操作は、バケットレベルで同時に実行できます。ただし、バケット数が多すぎると、多くの小さなファイルが作成される可能性があります。したがって、必要に応じてこの値を設定する必要があります。詳細については、「テーブル操作」および「データストレージとシャーディング」をご参照ください。
自動的に作成された Delta テーブルの履歴データのデフォルトのクエリ可能期間は 0 時間です。この期間は、宛先テーブルマッピングの [履歴データのクエリ可能期間] 設定で変更できます。
既存のテーブルのタイムトラベル保持期間は変更できません。この設定は、データの履歴バージョンをクエリできる時間範囲を決定します。指定された期間より古い履歴データはクエリできません。保持期間が長いほど、より多くの履歴データが保持され、ストレージコストが増加します。したがって、必要に応じてこの設定を構成する必要があります。詳細については、「テーブル操作」および「タイムトラベル」をご参照ください。
a. フィールドのデータ型マッピングを変更する
[宛先テーブルのマッピングルール] セクションの右上隅にある [フィールドデータ型のマッピングを編集] をクリックして、ビジネス要件に基づいてソースフィールドと宛先フィールド間のデータ型マッピングを構成できます。構成が完了したら、[適用してマッピングをリフレッシュ] をクリックします。
b. 宛先テーブルのスキーマを変更して、テーブルにフィールドを追加し、フィールドに値を割り当てる
宛先テーブルが [作成予定] 状態の場合、次の手順を実行してテーブルにフィールドを追加し、フィールドに値を割り当てることができます。
1 つ以上の宛先テーブルにフィールドを追加します。
単一の宛先テーブルにフィールドを追加: フィールドを追加する宛先テーブルを探し、[宛先テーブル名] 列の
アイコンをクリックします。表示されるダイアログボックスで、フィールドを追加します。一度に複数の宛先テーブルにフィールドを追加する: 一度にフィールドを追加する宛先テーブルを選択し、ページの下部にある [一括変更] をクリックし、[宛先テーブルスキーマ - フィールドの一括変更と追加] をクリックします。
フィールドに値を割り当てます。次のいずれかの操作を実行して、フィールドに値を割り当てることができます。
単一の宛先テーブルに追加されたフィールドに値を割り当てる: 新しく追加されたフィールドに値を割り当てる宛先テーブルを見つけ、[値の割り当て] 列の [構成] をクリックします。[追加フィールド] ダイアログボックスで、フィールドに値を割り当てます。
一度に複数の宛先テーブルに追加されたフィールドに値を割り当てる: 新しく追加されたフィールドに値を割り当てる宛先テーブルを選択し、ページの下部にある [一括変更] をクリックし、[値の割り当て] をクリックして、選択した宛先テーブルの同じフィールドに一度に値を割り当てます。
説明
アイコンをクリックして値の割り当て方法を切り替え、宛先テーブルに追加されたフィールドに定数と変数を割り当てることができます。
c. DML 処理ルールを構成する
Data Integration は、デフォルトの DML 処理ルールを提供します。ビジネス要件に基づいて、宛先テーブルの DML 処理ルールを構成することもできます。
単一の宛先テーブルの DML 処理ルールを構成する: DML 処理ルールを構成する宛先テーブルを見つけ、[DML ルールの構成] 列の [構成] をクリックして、テーブルの DML 処理ルールを構成します。
一度に複数の宛先テーブルの DML 処理ルールを構成する: DML 処理ルールを構成する宛先テーブルを選択し、ページの下部にある [一括変更] をクリックし、[DML ルールの構成] をクリックします。
4. ソースシャードキーを設定する
**[ソースシャーディング]** 列で、ソーステーブルからフィールドを選択するか、[シャーディングなし] を選択できます。
5. 完全同期を実行する
同期タスクタイプを選択したときに [同期ステップ] セクションで [完全同期] を選択した場合、このページで特定のテーブルの完全同期を無効にできます。
5. アラートルールを構成する
同期タスクの失敗によるビジネスデータ同期の遅延を防ぐために、同期タスクに異なるアラートルールを構成できます。
ページの右上隅にある [アラートルールの構成] をクリックして、[アラートルールの構成] パネルに移動します。
[アラートルールの構成] パネルで、[アラートルールの追加] をクリックします。[アラートルールの追加] ダイアログボックスで、パラメーターを構成してアラートルールを構成します。
説明このステップで構成するアラートルールは、同期タスクによって生成されるリアルタイム同期サブタスクに対して有効になります。同期タスクの設定が完了したら、「リアルタイム同期タスクの管理」を参照してリアルタイム同期タスクページに移動し、リアルタイム同期サブタスクに設定されたアラートルールを変更できます。
アラートルールを管理します。
作成されたアラートルールを有効または無効にできます。アラートの重大度レベルに基づいて、異なるアラート受信者を指定することもできます。
6. 詳細パラメーターを構成する
ビジネス要件に基づいて、同期タスクに構成された特定のパラメーターの値を変更できます。たとえば、現在の同期タスクがソースデータベースに過度の圧力をかけ、データ生成に影響が及ばないように、[最大読み取り接続数] パラメーターに適切な値を指定できます。
予期しないエラーやデータ品質の問題を防ぐために、パラメーターの値を変更する前に、パラメーターの意味を理解することをお勧めします。
設定ページの右上隅にある [詳細パラメーターの構成] をクリックします。
[詳細パラメーターの構成] パネルで、目的のパラメーターの値を変更します。
7. DDL 処理ルールを構成する
DDL 操作はソースで実行される場合があります。ページの右上隅にある [DDL 機能の構成] をクリックして、ビジネス要件に基づいてソースからの DDL メッセージを処理するルールを構成できます。
詳細については、「DDL メッセージを処理するルールを構成する」をご参照ください。
8. リソースグループの表示と変更
ページの右上隅にある [リソースグループの構成] をクリックして、現在の同期タスクの実行に使用されるリソースグループを表示および変更できます。
9. 同期タスクを実行する
同期タスクの設定が完了したら、ページの下部にある [完了] をクリックします。
[Data Integration] ページの [ノード] セクションで、作成した同期タスクを見つけ、[アクション] 列の [開始] をクリックします。
[タスク] セクションの同期タスクの [名前または ID] をクリックし、同期タスクの詳細な実行プロセスを表示します。
同期タスクの O&M
タスク実行ステータスの表示
同期タスクを作成した後、同期タスクページで同期タスクのリストとその基本情報を表示できます。
**[アクション]** 列で、同期タスクを [開始] または [停止] できます。**[その他]** メニューから [編集] や [表示] などの他の操作も実行できます。
実行中のタスクについては、[実行概要] セクションで基本的な操作ステータスを表示できます。概要のエリアをクリックして実行の詳細を表示することもできます。
MySQL データベース全体を MaxCompute に同期するリアルタイム同期タスクは、次の 3 つのステップで構成されます。
スキーマ移行: このステップには、宛先テーブルの作成方法が含まれます。これは、既存のテーブルまたは自動的に作成されたテーブルにすることができます。テーブルが自動的に作成される場合、DDL 文が表示されます。
完全初期化: このステップには、オフライン同期用のテーブル、同期の進行状況、および書き込まれたレコード数に関する情報が含まれます。
リアルタイムデータ同期: このステップには、リアルタイムの進行状況、DDL レコード、DML レコード、アラート情報などのリアルタイム同期統計が含まれます。
同期タスクの再実行
特別な場合に、ソースにテーブルを追加したり、ソースからテーブルを削除したり、宛先テーブルのスキーマや名前を変更したりした場合は、同期タスクの [アクション] 列の [その他] をクリックし、[再実行] をクリックして変更後にタスクを再実行できます。再実行プロセス中、同期タスクは、新しく追加されたテーブルから宛先へ、またはマッピングされたソーステーブルからスキーマまたは名前が変更された宛先テーブルへのみデータを同期します。
タスクの設定を変更せずに同期タスクを再実行する場合は、[アクション] 列の [その他] をクリックし、[再実行] をクリックしてタスクを再実行し、完全同期と増分同期を再度実行します。
タスクにテーブルを追加またはタスクからテーブルを削除した後に同期タスクを再実行する場合は、変更後に [完了] をクリックします。この場合、同期タスクの [アクション] 列に [更新の適用] が表示されます。[更新の適用] をクリックして、システムに同期タスクの再実行をトリガーさせます。再実行プロセス中、同期タスクは新しく追加されたテーブルから宛先にデータを同期します。元のテーブルのデータは再度同期されません。