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

前提条件

• RDS for MySQL インスタンスを作成しておきます。 詳細については、「RDS for MySQL インスタンスの作成」をご参照ください。 このトピックでは、MySQL 5.7 を例として使用します。

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

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

課金

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

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

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

RDS for MySQL インスタンスでデータベースとテーブルを作成します。

• Alibaba Cloud RDS データベース、またはローカルサーバー上のセルフマネージドデータベースを使用できます。 このトピックでは、RDS for MySQL データベースを例に説明します。 詳細については、「クイックスタート」をご参照ください。

• 次のサンプルコードは、テーブルを作成してデータを挿入する方法を示しています。

  -- テーブルの作成
  CREATE TABLE `es_test` (
      `id` bigint(32) NOT NULL,
      `name` varchar(32) NULL,
      `age` bigint(32) NULL,
      `hobby` varchar(32) NULL,
      PRIMARY KEY (`id`)
  ) ENGINE=InnoDB
  DEFAULT CHARACTER SET=utf8;
  
  -- データの挿入
  INSERT INTO `es_test` (`id`,`name`,`age`,`hobby`) VALUES (1,'user1',22,'music');
  INSERT INTO `es_test` (`id`,`name`,`age`,`hobby`) VALUES (2,'user2',23,'sport');
  INSERT INTO `es_test` (`id`,`name`,`age`,`hobby`) VALUES (3,'user3',43,'game');
  INSERT INTO `es_test` (`id`,`name`,`age`,`hobby`) VALUES (4,'user4',24,'run');
  INSERT INTO `es_test` (`id`,`name`,`age`,`hobby`) VALUES (5,'user5',42,'basketball');

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

Data Integration リソースグループを購入し、VPC とワークスペースをリソースグループに関連付けます。

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

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

3. リソースグループの作成 をクリックし、プロンプトに従ってパラメーターを設定します。

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

   データ同期を可能にするには、リソースグループが RDS for MySQL インスタンスと Elasticsearch インスタンスの両方に接続できる必要があります。 したがって、リソースグループをインスタンスと同じ [Virtual Private Cloud (VPC)]、Zone、vSwitch に関連付ける必要があります。 インスタンスの VPC 情報を確認するには、「RDS for MySQL インスタンスの VPC と vSwitch の切り替え」および「Elasticsearch インスタンスの基本情報の表示」をご参照ください。

   重要

   vSwitch CIDR ブロック の CIDR ブロックを、RDS for MySQL インスタンスと Elasticsearch インスタンスの両方の IP ホワイトリストに追加する必要があります。 詳細については、「RDS for MySQL インスタンスの IP ホワイトリストの設定」および「Elasticsearch インスタンスのパブリックまたはプライベート IP ホワイトリストの設定」をご参照ください。

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

6. 作成したリソースグループの Actions 列で、[ワークスペースのバインド] をクリックして、ターゲットワークスペースをリソースグループに関連付けます。

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

RDS for MySQL と Elasticsearch のデータソースを DataWorks の Data Integration サービスに追加します。

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

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

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

   3. ターゲットワークスペースの Actions 列で、[クイックアクセス] > Data Integration の順に選択します。

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

3. RDS for MySQL データソースを追加します。

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

   2. ソースインスタンスの追加 ページで、MySQL を検索して選択します。

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

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

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

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

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

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

オフライン同期タスクは、コンピューティングリソースとして機能するリソースグループで実行されます。 リソースグループは、設定されたデータソースからデータを取得し、Elasticsearch に書き込みます。

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

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

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

   3. ターゲットワークスペースの Actions 列で、[クイックアクセス] > データ開発 の順に選択します。

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

   1. 左側のナビゲーションペインの Data Studio タブ ( アイコン) で、新規作成 > 業務フローの作成 の順に選択します。 次に、プロンプトに従ってワークフローを作成します。

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

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

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

   1. データソース エリアの データソース で MySQL を選択し、ソースインスタンス名 で同期するデータソースの名前を選択します。

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

   3. データ宛先 セクションの データ宛先 で Elasticsearch を選択し、ソースインスタンス名 で同期するデータソースの名前を選択します。

4. Next step をクリックします。

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

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

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

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

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

   設定の詳細については、「ウィザードを使用した同期タスクの設定」をご参照ください。

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

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

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

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

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

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

ステップ5：結果の検証

1. 宛先の Alibaba Cloud Elasticsearch インスタンスの Kibana コンソールにログインします。 詳細については、「Kibana コンソールへのログイン」をご参照ください。

2. Kibana ページの左上隅にあるアイコンをクリックし、[開発ツール] を選択します。

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

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

   説明

   es_test を、データ同期タスクで指定したインデックス名に置き換えます。

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

   POST /es_test/_search?pretty
   
   {
     "took" : 0,
     "timed_out" : false,
     "_shards" : {
       "total" : 1,
       "successful" : 1,
       "skipped" : 0,
       "failed" : 0
     },
     "hits" : {
       "total" : 5,
       "max_score" : 1.0,
       "hits" : [
         {
           "_index" : "es_test",
           "_type" : "es_test",
           "_id" : "2",
           "_score" : 1.0,
           "_source" : {
             "age" : 23,
             "hobby" : "sport",
             "id" : 2,
             "name" : "user2"
           }
         },
         {
           "_index" : "es_test",
           "_type" : "es_test",
           "_id" : "1",
           "_score" : 1.0,
           "_source" : {
             "age" : 22,
             "hobby" : "music",
             "id" : 1,
             "name" : "user1"
           }
         }
       ]
     }
   }