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

前提条件

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

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

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

課金

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

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

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

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

• ApsaraDB for RDS データベースまたはローカルサーバー上の自己管理データベースを使用できます。このトピックでは、ApsaraDB 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. 作成したリソースグループの [操作] 列で、[ネットワーク設定] をクリックして、リソースグループに VPC をアタッチします。詳細については、「ネットワーク設定」をご参照ください。

   リソースグループは、データを同期するために ApsaraDB RDS for MySQL および ES インスタンスの VPC に接続できる必要があります。[VPC]、[ゾーン]、および [VSwitch] を、ApsaraDB RDS for MySQL および ES インスタンスが属するものにアタッチする必要があります。インスタンスの VPC 情報を表示するには、「ApsaraDB RDS for MySQL インスタンスの VPC と vSwitch の切り替え」および「Elasticsearch インスタンスの基本情報の表示」をご参照ください。

   重要

   VPC をアタッチした後、VPC の [VSwitch CIDR ブロック] を ApsaraDB RDS for MySQL および ES インスタンスの内部向けホワイトリストに追加する必要があります。詳細については、「ApsaraDB RDS for MySQL インスタンスの IP アドレスホワイトリストの設定」および「ES インスタンスのパブリックまたは内部向けアクセスホワイトリストの設定」をご参照ください。

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

6. 作成したリソースグループの [操作] 列で、[ワークスペースのアタッチ] をクリックして、ターゲットワークスペースをリソースグループにアタッチします。

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

   3. [フィールドマッピング] セクションで、[ソースフィールド] と [宛先フィールド] の間のマッピング関係を設定します。

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

   設定の詳細については、「コードレス UI での同期ノードの設定」をご参照ください。

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

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

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

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

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

      ログに Shell run successfully! が含まれている場合、タスクが正常に実行されたことを示します。

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

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

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

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

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

   説明

   es_test をデータ同期タスクで指定したインデックス名に置き換えてください。

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