このトピックでは、DataWorks Data Integration を使用して、MySQL データベースのすべてのデータをリアルタイムで MaxCompute Delta テーブルに同期する方法について説明します。
前提条件
MaxCompute データソースと MySQL データソースが準備されている。使用するデータソースを事前に DataWorks に追加する必要があります。これにより、同期タスクを構成するときに、データソースを選択して、データの読み取り元と書き込み先を決定できます。 MySQL データソースと MaxCompute データソースを追加する方法については、MySQL データソースとMaxCompute データソースの追加を参照してください。
説明データソースを追加する前に理解しておく必要がある項目については、データソースの追加と管理を参照してください。
ビジネス要件を満たすサーバーレスリソースグループが購入されている。
サーバーレスリソースグループとデータソース間のネットワーク接続が確立されている。詳細については、ネットワーク接続ソリューションを参照してください。
データソース環境が準備されている。
MaxCompute: MaxCompute V2.0 データ型エディションは DECIMAL データ型をサポートしています。 MaxCompute データソースを使用する同期タスクを実行する前に、データソースに対して MaxCompute V2.0 データ型エディションを有効にする必要があります。詳細については、MaxCompute V2.0 データ型エディションを参照してください。
MySQL: MySQL データソースに対してバイナリロギング機能を有効にする必要があります。詳細については、MySQL データソースを参照してください。
背景情報
DataWorks Data Integration を使用すると、リアルタイム同期タスクを実行して、MySQL データベースのすべてのデータを一度に同期し、MySQL データベースの増分データをリアルタイムで MaxCompute のパーティション分割された Delta テーブルまたはパーティション分割されていない Delta テーブルに同期できます。 MaxCompute Delta テーブルに同期されるデータは、分単位のほぼリアルタイムの可視性を備えています。増分同期が完了すると、少なくとも 5 分のレイテンシで MaxCompute Delta テーブルに同期された増分データをクエリできます。
MaxCompute Delta テーブルについては、ニアリアルタイムデータウェアハウスの概要を参照してください。
注意事項
MySQL データベースのすべてのデータをリアルタイムで MaxCompute に同期することは、テナントレベルのスキーマシнтаксисが有効になっている MaxCompute データソースではサポートされていません。
手順
手順
手順 1: 同期タイプの選択
Data Integration ページに移動します。
DataWorks コンソールにログインします。上部のナビゲーションバーで、目的のリージョンを選択します。左側のナビゲーションペインで、 を選択します。表示されたページで、ドロップダウンリストから目的のワークスペースを選択し、[データ統合に移動] をクリックします。
Data Integration ページの左側のナビゲーションペインで、[同期タスク] をクリックします。表示されるページの上部で、[ソース] ドロップダウンリストから [mysql] を選択し、[転送先] ドロップダウンリストから [maxcompute] を選択し、[作成] をクリックします。[データ同期ソリューションの作成] ページの [基本設定] セクションで、次のパラメーターを構成します。
[新しいノード名]: ビジネス要件に基づいて同期タスクの名前を指定します。
[同期方法]: [データベース全体のリアルタイム移行] を選択します。
手順 2: ネットワーク接続の確立
[ネットワークとリソースの構成] セクションで、ソースとして MySQL データソース、転送先として MaxCompute データソース、そしてサーバーレスリソースグループを選択します。次に、[すべてのリソースグループとデータソースの接続テスト] をクリックして、サーバーレスリソースグループとデータソース間のネットワーク接続をテストします。
[次へ] をクリックします。
手順 3: データを同期するテーブルの選択
この手順では、ソース テーブル
選択されたテーブル リストでデータを同期するテーブルを選択し、 アイコンをクリックして、選択したテーブルを リストに移動できます。また、正規表現に基づいてソースデータベースとソーステーブルをフィルタリングすることもできます。
手順 4: 転送先テーブルに関連する設定の構成
データを同期するテーブルを選択すると、選択したテーブルが [転送先テーブルのマッピングルール] セクションに自動的に表示されます。転送先テーブルのプロパティはマッピングされるのを待っています。ソーステーブルと転送先テーブル間のマッピングを手動で定義して、データの読み取りと書き込みの関係を決定する必要があります。その後、[アクション] 列の [更新] をクリックできます。ソーステーブルと転送先テーブル間のマッピングを直接更新できます。転送先テーブルに関連する設定を構成した後に、ソーステーブルと転送先テーブル間のマッピングを更新することもできます。
ソーステーブルが表示されている行を選択し、[マッピング結果の一括更新] をクリックできます。「転送先テーブル名のマッピングルールをカスタマイズする」パラメーターが構成されていない場合、関連する転送先テーブルには
${ソースデータベース名}_${ソーステーブル名}形式のマッピングルールが使用されます。転送先にソーステーブルと同じ名前のテーブルが存在しない場合、システムは転送先にテーブルを自動的に作成します。[転送先テーブル名のマッピングルールをカスタマイズする] 列の [編集] をクリックして、ビジネス要件に基づいて転送先テーブル名のマッピングルールを構成できます。
組み込み変数と特定の文字列を連結して、転送先テーブル名にすることができます。組み込み変数を編集できます。たとえば、マッピングルールを作成するときに、ソーステーブル名を示す変数に接尾辞を追加して、転送先テーブル名を作成できます。
MaxCompute Delta テーブルにデータを同期する場合、テーブルのプライマリキーを指定する必要があります。デフォルトでは、MaxCompute Delta テーブルは、マッピングされたソーステーブルと同じプライマリキーを使用します。ソーステーブルにプライマリキーがない場合は、ビジネス要件に基づいて MaxCompute Delta テーブルのプライマリキー列を構成できます。 MaxCompute Delta テーブルにプライマリキーが指定されていない場合、同期タスクを保存できません。
デフォルトでは、自動的に作成される MaxCompute Delta テーブルのバケット数は 16 です。[テーブルバケット番号] 列で数値を変更できます。
既存の MaxCompute Delta テーブルのデフォルトのバケット数を変更することはできません。データはテーブルのバケットに分散されます。データクエリ、データ書き込み、データマージなどの操作は、バケットの粒度で並列して実行できます。多数のバケットを指定すると、過剰な小さなファイルが生成されます。ビジネス要件に基づいてバケット数を指定することをお勧めします。詳細については、テーブル操作とバケットを参照してください。
デフォルトでは、自動的に作成される MaxCompute Delta テーブルの履歴データをクエリすることはできません。[ACID データ保持時間] 列の時間間隔を変更して、指定された時間内に生成された履歴データのクエリを有効にすることができます。
既存の MaxCompute Delta テーブルの時間間隔を変更することはできません。時間間隔外の履歴データはクエリできません。長い時間間隔を指定すると、より多くの履歴データが保持され、ストレージ料金が高くなります。ビジネス要件に基づいて時間間隔を指定することをお勧めします。詳細については、テーブル操作とタイムトラベルを参照してください。
データ型マッピングの変更
[転送先テーブルのマッピングルール] セクションの右上隅にある [フィールドデータ型のマッピングを編集] をクリックして、ビジネス要件に基づいてソースフィールドと転送先フィールド間のデータ型マッピングを構成できます。構成が完了したら、[適用してマッピングを更新] をクリックします。
転送先テーブルのスキーマを変更してテーブルにフィールドを追加し、フィールドに値を割り当てる
MaxCompute Delta テーブルが [作成予定] 状態の場合、次の手順を実行してテーブルにフィールドを追加し、フィールドに値を割り当てることができます。
1 つ以上の転送先テーブルにフィールドを追加します。
単一の転送先テーブルにフィールドを追加する: フィールドを追加する転送先テーブルを見つけ、
[転送先テーブル名] 列の アイコンをクリックします。表示されるダイアログボックスで、フィールドを追加します。複数の転送先テーブルに一度にフィールドを追加する: 一度にフィールドを追加する転送先テーブルを選択し、ページの下部にある [一括変更] をクリックし、[転送先テーブルスキーマ - 一括変更とフィールドの追加] をクリックします。
フィールドに値を割り当てます。次のいずれかの操作を実行して、フィールドに値を割り当てることができます。
単一の転送先テーブルに追加されたフィールドに値を割り当てる: 新しく追加されたフィールドに値を割り当てる転送先テーブルを見つけ、[値の割り当て] 列の [構成] をクリックします。[追加フィールド] ダイアログボックスで、フィールドに値を割り当てます。
複数の転送先テーブルに追加されたフィールドに一度に値を割り当てる: 新しく追加されたフィールドに値を割り当てる転送先テーブルを選択し、ページの下部にある [一括変更] をクリックし、[値の割り当て] をクリックして、選択した転送先テーブルの同じフィールドに一度に値を割り当てます。
説明
アイコンをクリックして、値の割り当て方法を切り替えることができます。転送先テーブルに追加されたフィールドに定数と変数を割り当てることができます。
DML 処理ルールの構成
Data Integration はデフォルトの DML 処理ルールを提供します。ビジネス要件に基づいて、転送先テーブルの DML 処理ルールを構成することもできます。
単一の転送先テーブルの DML 処理ルールを構成する: DML 処理ルールを構成する転送先テーブルを見つけ、[DML ルールの構成] 列の [構成] をクリックして、テーブルの DML 処理ルールを構成します。
複数の転送先テーブルの DML 処理ルールを一度に構成する: DML 処理ルールを構成する転送先テーブルを選択し、ページの下部にある [一括変更] をクリックし、[DML ルールの構成] をクリックします。
手順 5: アラートルールを構成する
同期タスクの失敗によってビジネスデータの同期に遅延が発生するのを防ぐために、同期タスクによって生成されるリアルタイム同期サブタスクに異なるアラートルールを構成できます。
ページの右上隅にある [アラートルールの構成] をクリックして、[アラートルールの構成] パネルに移動します。
[アラートルールの構成] パネルで、[アラートルールの追加] をクリックします。[アラートルールの追加] ダイアログボックスで、パラメーターを構成してアラートルールを構成します。
説明この手順で構成するアラートルールは、同期タスクによって生成されるリアルタイム同期サブタスクに有効になります。同期タスクの構成が完了したら、リアルタイム同期タスクの管理を参照して [リアルタイム同期タスク] ページに移動し、リアルタイム同期サブタスクに構成されたアラートルールを変更できます。
アラートルールを管理します。作成されたアラートルールを有効または無効にすることができます。アラートの重大度レベルに基づいて、異なるアラート受信者を指定することもできます。
手順 6: 詳細パラメーターを構成する
Data Integration は、「最大読み取り接続数」や「タスクの同時実行性」などのパラメーターのデフォルト値を提供します。ビジネス要件に合わせてきめ細かい構成を行う場合は、パラメーターの値を変更できます。たとえば、「最大読み取り接続数」パラメーターに適切な値を指定して、現在の同期タスクがソースデータベースに過度の負荷をかけたり、データ生成に影響を与えたりするのを防ぐことができます。
予期しないエラーやデータ品質の問題を防ぐために、パラメーターの値を変更する前に、パラメーターの意味を理解することをお勧めします。
ページの右上隅にある 詳細パラメーターの構成 をクリックします。
[詳細設定パラメーターの構成] パネルで、ビジネス要件とリソースグループおよびデータベースの状況に基づいて、次の表に示すパラメーターを構成します。
タブ
パラメーター
説明
[リーダー設定]
[最大読み取り接続数]
現在の同期タスクがソースからデータを読み取るために使用できる並列スレッドの最大数。このパラメーターを構成して、現在の同期タスクが占有できるソースデータベース接続の数を制御できます。
[ライター設定]
[リアルタイムタスクセッションキャッシュサイズ]
リアルタイム同期サブタスクのセッションのキャッシュサイズ。単位: バイト。このパラメーターの値を超えると、データは MaxCompute にフラッシュされます。このパラメーターを構成しない場合、MaxCompute 側のデフォルト値が使用されます。
[リアルタイムタスクバケットキャッシュサイズ]
リアルタイム同期サブタスクのバケットのキャッシュサイズ。単位: バイト。このパラメーターの値を超えると、データは MaxCompute にフラッシュされます。このパラメーターを構成しない場合、MaxCompute 側のデフォルト値が使用されます。
[ランタイム設定]
[タスクの同時実行性]
現在の同期タスクに使用できる並列スレッドの数。
[同時コミットスレッド数]
同期タスクのコミットに使用できる並列スレッドの数。このパラメーターを 1 に設定すると、同期タスクのコミットに並列スレッドは使用されません。デフォルト値: 5。
手順 7: DDL 処理ルールの構成
ソースで DDL 操作が実行される場合があります。ページの右上隅にある [DDL 機能の構成] をクリックして、ビジネス要件に基づいてソースからの DDL メッセージを処理するルールを構成できます。
詳細については、DDL メッセージを処理するためのルールの構成を参照してください。
手順 8: リソースグループの構成
ページの右上隅にある [リソースグループの構成] をクリックして、現在の同期タスクの実行に使用するリソースグループを表示および変更できます。
手順 9: 同期タスクの実行
同期タスクの構成が完了したら、ページの下部にある [完了] をクリックします。
[データ統合] ページの [ノード] セクションで、作成された同期タスクを見つけ、[アクション] 列の [開始] をクリックします。
[ノード] セクションで同期タスクの [名前または ID] をクリックし、同期タスクの実行の詳細なプロセスを表示します。
次の手順
同期タスクの構成が完了したら、同期タスクの管理、同期タスクへのテーブルの追加または削除、同期タスクの監視およびアラート設定の構成、同期タスクの実行に関連する主要メトリックの表示を行うことができます。詳細については、フル同期タスクと増分同期タスクの O&M を実行するを参照してください。