Elasticsearch:DataWorks を使用して HBase から Alibaba Cloud ES にデータを同期する

前提条件

• ApsaraDB for HBase クラスターが作成されていること。詳細については、「クラスターの購入」をご参照ください。

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

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

課金

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

• データ統合専用リソースグループの課金の詳細については、「データ統合専用リソースグループの課金 (サブスクリプション)」をご参照ください。

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

この例では、次のテーブル作成ステートメントとテストデータを使用します。HBase クラスターにデータをインポートする方法の詳細については、「HBase Shell を使用してアクセスする」をご参照ください。

• テーブル作成ステートメント

  create 'student', {NAME => 'name'}, {NAME => 'ID'}, {NAME => 'gender'}

• テストデータ

  put コマンドを使用して、テーブルにデータを挿入します。例: put 'student', 'row1', 'name:a', 'xiaoming'。

  scan コマンドを使用して、テーブル内のデータを表示します。例: scan 'student'。

ステップ 2: 専用リソースグループの購入と設定

データ統合専用リソースグループを購入し、VPC とワークスペースをリソースグループにアタッチします。専用リソースグループは、高速で安定したデータ転送を保証します。

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

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

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

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

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

5. 作成した専用リソースグループの [操作] 列で、[ネットワーク設定] をクリックして VPC をアタッチします。

   説明

   この例では、データ統合専用リソースグループを使用して VPC 経由でデータを同期します。詳細については、「IP アドレスホワイトリストの設定」をご参照ください。

   データを同期するには、専用リソースを HBase インスタンスと Elasticsearch インスタンスの VPC に接続する必要があります。したがって、専用リソースを HBase インスタンスと Elasticsearch インスタンス両方の [VPC]、[ゾーン]、および [vSwitch] にアタッチする必要があります。ES インスタンスの VPC の詳細については、「Elasticsearch インスタンスの基本情報を表示する」をご参照ください。

   重要

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

6. 左上隅にある戻るアイコンをクリックして、[リソースグループリスト] ページに戻ります。

7. 専用リソースグループの [操作] 列で、[ワークスペースのアタッチ] をクリックして、宛先ワークスペースをアタッチします。

   詳細については、「ステップ 2: リソースグループをワークスペースにバインドする」をご参照ください。

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

HBase と Elasticsearch のデータソースを DataWorks のデータ統合に追加します。

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

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

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

   3. 宛先ワークスペースの [操作] 列で、[移動] > [データ統合] を選択します。

2. データ統合ページの左側のナビゲーションウィンドウで、[データソース] をクリックします。

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

   1. [データソースリスト] ページで、[データソースの追加] をクリックします。

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

   3. [HBase データソースの追加] ダイアログボックスで、[基本情報] セクションのデータソースパラメーターを設定できます。

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

   4. [接続設定] セクションで、[接続テスト] をクリックします。[接続済み] のステータスは、接続が成功したことを示します。

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

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

ステップ 4: オフライン同期タスクの設定と実行

オフライン同期タスクは、専用リソースグループを使用して実行されます。専用リソースグループは、データ統合のデータソースからデータを取得し、ES インスタンスにデータを書き込みます。

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

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

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

   3. 宛先ワークスペースの [操作] 列で、[移動] [データ開発] を選択します。

2. オフライン同期タスクを作成します。

   1. 左側のナビゲーションウィンドウのデータ開発 ( アイコン) タブで、[作成] > [ワークフローの作成] を選択し、画面の指示に従ってワークフローを作成します。

   2. 作成したワークフローを右クリックし、[ノードの作成] > [データ統合] > [オフライン同期] を選択します。

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

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

   1. [ソース] セクションで、[ソース] を HBase に、[データソース名] をデータソースの名前に設定します。

   2. [リソースグループ] で、専用リソースグループを選択します。

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

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

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

   1. [ソース] セクションで、同期するテーブルを選択します。

   2. [宛先] セクションでパラメーターを設定できます。

   3. [フィールドマッピング] セクションで、[ソースフィールド] と [宛先フィールド] の間のマッピングを設定します。詳細については、「コードレス UI を使用してオフライン同期タスクを設定する」をご参照ください。

   4. [チャネルコントロール] セクションで、チャネルパラメーターを設定できます。

   詳細については、「コードレス UI を使用してバッチ同期タスクを設定する」をご参照ください。

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

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

   2. ノードエディターの上にある保存アイコンをクリックして、タスクを保存します。

   3. ノードエディターの上にある送信アイコンをクリックして、タスクを送信します。

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

      運用ログに Shell run successfully! と表示された場合、タスクは正常に実行されています。

ステップ 5: データ同期結果の確認

1. Elasticsearch クラスターの Kibana コンソールにログインし、プロンプトに従って Kibana コンソールのホームページに移動します。

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

   説明

   この例では、Elasticsearch V7.10.0 クラスターを使用します。他のバージョンのクラスターでの操作は異なる場合があります。コンソールでの実際の操作が優先されます。

2. 表示されたページの右上隅で、[Dev Tools] をクリックします。

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

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

   説明

   student_info は、オフライン同期タスクで宛先に設定したインデックス名です。