Elasticsearch:DataWorks を使用した PolarDB-X データの Alibaba Cloud Elasticsearch への同期

前提条件

• PolarDB-X インスタンスが作成済みである必要があります。詳細については、「」をご参照ください。

• Alibaba Cloud Elasticsearch インスタンスが作成済みであり、自動インデックス作成機能が有効になっている必要があります。詳細については、「Alibaba Cloud Elasticsearch インスタンスの作成」および「YML ファイルの構成」をご参照ください。

• DataWorks ワークスペースが作成されています。詳細については、「ワークスペースを作成する」をご参照ください。

課金

• Alibaba Cloud Elasticsearch インスタンスの料金に関する情報は、「Elasticsearch 課金項目」をご参照ください。

• Data Integration リソースグループの料金に関する情報は、「リソースグループ料金」をご参照ください。

ステップ 1：ソースデータの準備

1. PolarDB-X 1.0 インスタンスにデータを挿入します。

   詳細については、「基本的な SQL 操作」をご参照ください。本トピックでは、次の図に示すテストデータを使用します。

ステップ 2：専用リソースグループの作成

Data Integration の排他的リソースグループを購入し、ご利用の VPC およびワークスペースに関連付けることで、高速かつ安定したデータ転送を実現します。

1. DataWorks コンソールにログインします。

2. 上部メニューバーでリージョンを選択し、左側ナビゲーションウィンドウで リソースグループ をクリックします。

3. 専用リソースグループ タブで、DI 用リソースグループの作成 > Data Integration リソースグループ をクリックします。

4. DataWorks 排他的リソース （サブスクリプション） の購入ページで、排他的 リソースタイプ を データ統合専用リソースグループ に設定し、リソースグループ名を入力して、今すぐ購入 をクリックして排他的リソースグループを購入します。

   詳細な構成情報については、「ステップ 1：リソースグループの購入」をご参照ください。

5. 作成した専用リソースグループの Actions 列で、ネットワーク設定 をクリックして、専用リソースグループを VPC に関連付けます。詳細については、「VPC の関連付け」をご参照ください。

   説明

   本トピックでは、データ統合専用リソースグループを使用して VPC 内部ネットワーク経由でデータを同期する方法について説明します。インターネット経由でのデータ同期については、「ホワイトリストの追加」をご参照ください。

   データを同期するには、排他的リソースグループが PolarDB-X および Elasticsearch インスタンスの VPC に接続されている必要があります。排他的リソースグループを PolarDB-X および Elasticsearch インスタンスの Virtual Private Cloud (VPC)、Zone、および vSwitch に関連付けます。インスタンスの VPC 情報を確認する方法については、「Elasticsearch インスタンスの基本情報の確認」をご参照ください。

   重要

   排他的リソースグループを VPC に関連付けた後、vSwitch CIDR ブロック の CIDR ブロックを PolarDB-X および Elasticsearch インスタンスの非公開アクセスホワイトリストに追加する必要があります。詳細については、「Elasticsearch インスタンスのパブリックまたは非公開アクセスホワイトリストの構成」をご参照ください。

6. ページ左上の戻るアイコンをクリックして、Resource List ページに戻ります。

7. 作成した排他的リソースグループの Actions 列で、ワークスペースの関連付け をクリックして、排他的リソースグループを宛先ワークスペースに関連付けます。

   詳細については、「ステップ 2：リソースグループのワークスペースへの関連付け」をご参照ください。

ステップ 3：データソースの追加

DataWorks Data Integration で PolarDB-X および Elasticsearch をデータソースとして追加します。

1. Data Integration ページに移動します。

   1. DataWorks コンソールにログインします。

   2. 左側ナビゲーションウィンドウで ワークスペース をクリックします。

   3. 対象のワークスペースを見つけ、Actions 列で ショートカット > Data Integration を選択します。

2. 左側ナビゲーションウィンドウで データソース をクリックします。

3. PolarDB-X データソースを追加します。

4. ソースインスタンス ページで、ソースインスタンスの追加 をクリックします。

5. データソースの追加 ページで、DRDS を検索して選択します。

6. DRDS データソースの追加 ページで、データソース情報を構成し、接続性をテストします。接続性テストが成功したら、完了 をクリックします。

   詳細については、「DRDS データソースの追加」をご参照ください。

7. 同様の手順で Elasticsearch データソースを追加します。詳細については、「Elasticsearch データソースの構成」をご参照ください。

ステップ 4：同期タスクの構成と実行

バッチ同期タスクは排他的リソースグループ上で実行され、Data Integration からソースのデータを読み取り、Elasticsearch に書き込みます。

1. DataWorks の データ開発 ページに移動します。

   1. DataWorks コンソールにログインします。

   2. 左側ナビゲーションウィンドウで ワークスペース をクリックします。

   3. 宛先ワークスペースの Actions 列で、クイックアクセス > データ開発 を選択します。

2. バッチ同期タスクを作成します。

   1. データ開発ページで () アイコンをクリックし、新規作成 > 業務フローの作成 を選択し、プロンプトに従ってビジネスフローを作成します。

   2. 作成したビジネスフローを右クリックし、ノードの作成 > オフライン同期 を選択します。

   3. ノードの作成 ダイアログボックスで、ノード名を入力し、OK をクリックします。

3. ネットワークおよびリソースを構成します。

   1. データソース セクションで、データソース を DRDS に設定し、ソースインスタンス名 をソースデータソースに設定します。

   2. リソースグループ セクションで、排他的リソースグループを選択します。

   3. データ宛先 セクションで、データ宛先 を Elasticsearch に設定し、ソースインスタンス名 を宛先データソースに設定します。

4. 次へをクリックします。

5. タスクを構成します。

   1. データソース セクションで、ソーステーブルを選択します。

   2. データ宛先 セクションで、送信先パラメーターを構成します。

   3. フィールドマッピング セクションで、ソースフィールド を ターゲットフィールド にマッピングします。構成の詳細については、「ウィザードモードでのタスク構成」をご参照ください。

      本例では、ソースフィールド は変更せず、ターゲットフィールド のみを変更します。ターゲットフィールド の右側にある アイコンをクリックし、ダイアログボックスに次のフィールド定義を入力します。

      {"name":"Name","type":"text"}
      {"name":"Platform","type":"text"}
      {"name":"Year_of_Release","type":"date"}
      {"name":"Genre","type":"text"}
      {"name":"Publisher","type":"text"}
      {"name":"na_Sales","type":"float"}
      {"name":"EU_Sales","type":"float"}
      {"name":"JP_Sales","type":"float"}
      {"name":"Other_Sales","type":"float"}
      {"name":"Global_Sales","type":"float"}
      {"name":"Critic_Score","type":"long"}
      {"name":"Critic_Count","type":"long"}
      {"name":"User_Score","type":"float"}
      {"name":"User_Count","type":"long"}
      {"name":"Developer","type":"text"}
      {"name":"Rating","type":"text"}

   4. チャンネル制御 セクションで、チャンネルパラメーターを構成します。

   詳細な構成情報については、「ウィザードモードでのタスク構成」をご参照ください。

6. タスクを実行します。

   1. （任意）タスクのスケジューリングプロパティを構成します。ページ右側で スケジューリング設定 をクリックし、必要に応じてスケジューリングパラメーターを構成します。各パラメーターの詳細については、「スケジューリング構成」をご参照ください。

   2. ノード編集エリアのツールバーで保存アイコンをクリックして、タスクを保存します。

   3. ノード編集エリアのツールバーで送信アイコンをクリックして、タスクを送信します。

      タスクにスケジューリングプロパティを構成した場合、タスクは定期的に実行されます。また、ノードエリア右上隅の実行アイコンをクリックして、タスクを即時実行することもできます。

      実行ログに Shell run successfully! というメッセージが表示された場合、タスクは正常に実行されています。

ステップ 5：同期結果の検証

1. 宛先の Alibaba Cloud Elasticsearch インスタンスの Kibana コンソールにログインします。

   詳細については、「Kibana コンソールへのログイン」をご参照ください。

2. 左側ナビゲーションウィンドウで Dev Tools をクリックします。

3. コンソール で、次のコマンドを実行して宛先のデータ件数を照会します。

   説明

   ソースと宛先のデータ件数を比較して、すべてのデータが同期されたことを検証します。

   GET drdstest/_search
   {
     "query": {
       "match_all": {}
     }
   }

   コマンドが正常に実行された場合、次の結果が返されます。

4. 次のコマンドを実行して、指定フィールドでデータを検索します。

   GET drdstest/_search
   {
     "query": {
       "term": {
         "Publisher.keyword": {
           "value": "Nintendo"
         }
       }
     }
   }

   コマンドが正常に実行された場合、次の結果が返されます。