Elasticsearch:Hadoop データの Alibaba Cloud Elasticsearch への同期

前提条件

• Alibaba Cloud Elasticsearch クラスターを作成し、その自動インデックス作成機能を有効にしておくこと。詳細については、「Alibaba Cloud Elasticsearch クラスターの作成」および「YML パラメーターの設定」をご参照ください。

  説明

  データを同期できるのは Alibaba Cloud Elasticsearch クラスターのみです。セルフマネージドの Elasticsearch クラスターはサポートされていません。

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

課金

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

• Data Integration リソースグループの料金については、「リソースグループの料金」をご参照ください。

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

高速で安定したデータ転送を確保するために、Data Integration 用の専用リソースグループを購入し、ワークスペースおよび Virtual Private Cloud (VPC) に関連付けます。

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

2. 上部メニューでリージョンを選択します。左側メニューで、リソースグループ をクリックします。

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

4. [DataWorks 専用リソース (サブスクリプション)] 購入ページで、[専用リソースタイプ] を [Data Integration 用専用リソースグループ] に設定し、リソースグループの名前を入力してから、今すぐ購入 をクリックします。

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

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

   説明

   このトピックでは、Data Integration の専用リソースグループを使用して VPC 経でデータを同期する例を説明します。インターネット経由でデータを同期する方法については、「IP アドレスホワイトリストの設定」をご参照ください。

   専用リソースグループは、Hadoop と Elasticsearch の両クラスターの VPC に接続し、両クラスター間のデータ同期を可能にする必要があります。したがって、専用リソースグループを、Hadoop クラスターと Elasticsearch クラスター両方の [Virtual Private Cloud (VPC)]、Zone、および [vSwitch] に関連付ける必要があります。Elasticsearch クラスターの VPC、ゾーン、vSwitch を表示するには、「Elasticsearch クラスターの基本情報の表示」をご参照ください。

   重要

   専用リソースグループを VPC に関連付けた後、Hadoop クラスターと Elasticsearch クラスターの VPC プライベートアクセスホワイトリストに [vSwitch CIDR ブロック] を追加する必要があります。詳細については、「Elasticsearch クラスターのパブリックまたはプライベート IP アドレスホワイトリストの設定」をご参照ください。

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

7. 作成した専用リソースグループを見つけ、Actions 列の [ワークスペースの関連付け] をクリックして、リソースグループを宛先のワークスペースに関連付けます。

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

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

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

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

   2. 左側のナビゲーションペインで、ワークスペース をクリックします。

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

2. 左側メニューで、データソース をクリックします。

3. HDFS データソースを追加します。

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

   2. ソースインスタンスの追加 ダイアログボックスで、[HDFS] を検索して選択します。

   3. [HDFS データソースの追加] ページで、データソースパラメーターを設定します。

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

   4. 接続テスト をクリックします。到達可能 が表示された場合、リソースグループはデータソースに接続されています。

   5. 完了 をクリックします。

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

ステップ 3：バッチ同期タスクの実行

バッチ同期タスクは、専用リソースグループで実行されます。リソースグループは Data Integration のデータソースからデータを取得し、宛先の Elasticsearch クラスターにデータを書き込みます。

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

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

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

   3. 送信先ワークスペースの Actions 列で、ショートカット > データ開発 を選択します。

2. オフライン同期ノードを作成します。

   1. Data Development () タブで、新規作成 > 業務フローの作成 を選択します。

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

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

3. ネットワークとリソースを設定します。

   1. データソース セクションで、データソース を HDFS に、ソースインスタンス名 を同期したい HDFS データソースの名前に設定します。

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

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

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

5. タスクを設定します。

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

   2. データ宛先 セクションで、パラメーターを設定します。

   3. フィールドマッピング セクションで、ソースフィールド と ターゲットフィールド の間のマッピングを設定します。

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

   詳細については、「ウィザードモードでのバッチ同期タスクの設定」をご参照ください。

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

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

   2. ノード設定タブのツールバーで、保存アイコンをクリックします。

   3. ノード設定タブのツールバーで、コミットアイコンをクリックします。

      スケジューリングプロパティを設定した場合、タスクは定期的に実行されます。ツールバーの実行アイコンをクリックして、タスクをすぐに実行することもできます。

      実行ログに Shell run successfully! メッセージが表示されたら、タスクが正常に完了したことを示します。

ステップ 4：結果の検証

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

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

2. 左側メニューで、[開発ツール] をクリックします。

3. [コンソール] で、次のコマンドを実行して同期されたデータを表示します。

   POST /hive_esdoc_good_sale/_search?pretty
   {
   "query": { "match_all": {}}
   }

   説明

   hive_esdoc_good_sale は、データ同期スクリプトの index パラメーターに指定した値です。

   データが正常に同期されると、次のような結果が返されます。

   
   {
     "took" : 6,
     "timed_out" : false,
     "_shards" : {
       "total" : 5,
       "successful" : 5,
       "skipped" : 0,
       "failed" : 0
     },
     "hits" : {
       "total" : 4,
       "max_score" : 1.0,
       "hits" : [
         {
           "_index" : "hive_esdoc_good_sale",
           "_type" : "_doc",
           "_id" : "2018-08-21 00:00:00",
           "_score" : 1.0,
           "_source" : {
             "trans_num" : 3,
             "click_cnt" : 7,
             "category" : "Outerwear",
             "buyer_id" : "lilei",
             "trans_amount" : 500.6,
             "brand" : "Brand A"
           }
         },
         {
           "_index" : "hive_esdoc_good_sale",
           "_type" : "_doc",
           "_id" : "2018-08-23 00:00:00",
           "_score" : 1.0,
           "_source" : {
             "trans_num" : 5,
             "click_cnt" : 4,
             "category" : "Outerwear",
             "buyer_id" : "jimmy",
             "trans_amount" : 100.2,
             "brand" : "Brand E"
           }
         },
         {
           "_index" : "hive_esdoc_good_sale",
           "_type" : "_doc",
           "_id" : "2018-08-22 00:00:00",
           "_score" : 1.0,
           "_source" : {
             "trans_num" : 2,
             "click_cnt" : 3
           }
         }
       ]
     }
   }