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