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