ApsaraDB RDS for SQL Server から Alibaba Cloud Elasticsearch へのデータ同期 - Elasticsearch

ApsaraDB RDS for SQL Server のデータを検索、分析する必要がある場合は、Alibaba Cloud Elasticsearch を使用できます。このトピックでは、DataWorks の Data Integration サービスを使用して、データを Alibaba Cloud Elasticsearch にオフラインで迅速に同期する方法について説明します。

背景

DataWorks は、ビッグデータエンジンを基盤とするエンドツーエンドのビッグデータ開発およびガバナンスプラットフォームです。データ開発、タスクスケジューリング、データ管理などの機能が統合されています。 DataWorks の同期タスクを使用すると、さまざまなデータソースから Alibaba Cloud Elasticsearch にデータを迅速に同期できます。

サポートされているデータソース：
- Alibaba Cloud データベース (MySQL、PostgreSQL、SQL Server、MongoDB、HBase)
- Alibaba Cloud PolarDB-X (DRDS からアップグレード)
- Alibaba Cloud MaxCompute
- Alibaba Cloud OSS
- Alibaba Cloud Tablestore
- HDFS、Oracle、FTP、DB2、およびその他のサポートされているデータベースタイプのセルフマネージドバージョン
シナリオ：
- ビッグデータの Alibaba Cloud Elasticsearch へのオフライン同期。データベース全体または特定のテーブルのすべてのデータを同期できます。詳細については、「MySQL データベース全体の Elasticsearch へのオフライン同期」をご参照ください。
- ビッグデータの Alibaba Cloud Elasticsearch へのリアルタイム同期。この方法は、完全同期と増分同期の両方をサポートしています。詳細については、「MySQL データベース全体の Elasticsearch へのリアルタイム同期」をご参照ください。

前提条件

ApsaraDB RDS for SQL Server インスタンスを作成済みであること。このトピックでは、ApsaraDB RDS for SQL Server 2019 Standard Edition インスタンスを例として使用します。詳細については、「ApsaraDB RDS for SQL Server インスタンスのクイック作成と使用」をご参照ください。
Alibaba Cloud Elasticsearch インスタンスを作成し、自動インデックス作成機能を有効にしていること。詳細については、「Alibaba Cloud Elasticsearch インスタンスの作成」および「YML パラメーターの設定」をご参照ください。
DataWorks ワークスペースを作成済みであること。詳細については、「ワークスペースの作成」をご参照ください。

説明

データは Alibaba Cloud Elasticsearch インスタンスにのみ同期できます。セルフマネージドの Elasticsearch インスタンスはサポートされていません。
ApsaraDB RDS for SQL Server インスタンス、Alibaba Cloud Elasticsearch インスタンス、および DataWorks ワークスペースは、同じリージョンにある必要があります。
ApsaraDB RDS for SQL Server インスタンス、Alibaba Cloud Elasticsearch インスタンス、および DataWorks ワークスペースは、同じタイムゾーンにある必要があります。そうでない場合、タイムゾーンの不一致により、同期後に時間関連のデータにオフセットが生じる可能性があります。

課金

Alibaba Cloud Elasticsearch インスタンスの料金については、「Elasticsearch の課金項目」をご参照ください。
Data Integration リソースグループの料金については、「リソースグループの料金」をご参照ください。

手順

手順 1：ソースデータの準備

ApsaraDB RDS for SQL Server インスタンスでは、次のサンプルデータを使用します。

説明

このトピックのデータはテスト専用です。本番環境では使用しないでください。

テーブルの作成

CREATE TABLE students(
    id INT,
    name VARCHAR(20),
    age INT
)

データの挿入

INSERT INTO [students]( [id] , [name] , [age] )values('1', 'Xiaoming','21');
INSERT INTO [students]( [id] , [name] , [age] )values('2', 'Xiaozhang','22');
INSERT INTO [students]( [id] , [name] , [age] )values('3', 'Xiaowang','23');

手順 2：専用リソースグループの購入と設定

Data Integration 用の専用リソースグループを購入し、Virtual Private Cloud (VPC) とワークスペースをアタッチします。専用リソースグループは、高速で安定したデータ転送を保証します。

DataWorks コンソールにログインします。
上部メニューでリージョンを選択します。左側のナビゲーションペインで、リソースグループ をクリックします。
専用リソースグループ タブで、[レガシーリソースグループの作成] > Data Integration リソースグループ をクリックします。
[DataWorks 専用リソース (サブスクリプション)] 購入ページで、[専用] リソースタイプ を [データ統合用専用リソース] に設定し、リソースグループの名前を入力してから、今すぐ購入 をクリックします。

設定情報の詳細については、「手順 1：リソースグループの購入」をご参照ください。
新しく作成した専用リソースグループの Actions 列で、ネットワーク設定 をクリックして VPC をリソースグループにアタッチします。詳細については、「VPC のアタッチ」をご参照ください。

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

データを同期するには、専用リソースグループが ApsaraDB RDS for SQL Server および Alibaba Cloud Elasticsearch インスタンスの VPC に接続できる必要があります。このため、ApsaraDB RDS for SQL Server および Alibaba Cloud Elasticsearch インスタンスの Virtual Private Cloud (VPC)、Zone、および vSwitch をアタッチする必要があります。インスタンスの VPC 情報を表示するには、「ApsaraDB RDS for SQL Server インスタンスの vSwitch を切り替える」および「Elasticsearch インスタンスの基本情報を表示する」をご参照ください。

重要
VPC をアタッチした後、vSwitch CIDR ブロック を、ApsaraDB RDS for SQL Server と Alibaba Cloud Elasticsearch 両インスタンスのプライベートホワイトリストに追加する必要があります。詳細については、「ApsaraDB RDS for SQL Server インスタンスのホワイトリストの設定」および「Elasticsearch インスタンスのパブリックまたはプライベートアクセスホワイトリストの設定」をご参照ください。
ページの左上隅にある戻るアイコンをクリックして、Resource List ページに戻ります。
新しく作成した専用リソースグループの Actions 列で、[ワークスペースのアタッチ] をクリックして、ターゲットワークスペースをリソースグループにアタッチします。

詳細については、「手順 2：ワークスペースのアタッチ」をご参照ください。

手順 3：データソースの追加

ApsaraDB RDS for SQL Server と Alibaba Cloud Elasticsearch のデータソースを DataWorks の Data Integration へ追加します。

DataWorks の Data Integration ページに移動します。
1. DataWorks コンソールにログインします。
2. 左側のナビゲーションペインで、ワークスペース をクリックします。
3. ターゲットワークスペースの Actions 列で、[クイックアクセス] > Data Integration を選択します。
左側のナビゲーションペインで、データソース をクリックします。
ApsaraDB RDS for SQL Server データソースを追加します。
1. ソースインスタンス ページで、ソースインスタンスの追加 をクリックします。
2. ソースインスタンスの追加 ページで、SQLServer データソースタイプを検索して選択します。
3. [SQL Server データソースの追加] ダイアログボックスで、基本情報 セクションのパラメーターを設定します。
  
  パラメーターの詳細については、「SQLServer データソースの設定」をご参照ください。
4. 接続設定 セクションで、接続テスト をクリックします。ステータスが 到達可能 と表示されれば、接続は成功です。
5. 完了をクリックします。
同じ手順に従って、Elasticsearch データソースを追加します。パラメーターの詳細については、「Elasticsearch データソースの設定」をご参照ください。

手順 4：オフライン同期タスクの設定と実行

オフライン同期タスクは専用リソースグループ上で実行されます。このリソースグループは、Data Integration で設定したソースからデータを取得し、Alibaba Cloud Elasticsearch にデータを書き込みます。

説明

オフライン同期タスクは、ウィザードモードまたはスクリプトモードで設定できます。このトピックでは、ウィザードモードでタスクを設定する方法について説明します。スクリプトモードについては、「スクリプトモードでの同期タスクの設定」および「Elasticsearch Writer」をご参照ください。

DataWorks の データ開発 ページに移動します。
1. DataWorks コンソールにログインします。
2. 左側のナビゲーションペインで、ワークスペース をクリックします。
3. ターゲットワークスペースの Actions 列で、[クイックアクセス] > データ開発 を選択します。
オフライン同期タスクを作成します。
1. 左側のナビゲーションペインで、アイコンをクリックし、新規作成 > 業務フローの作成 を選択して、ビジネスフローを作成します。
2. 新しいビジネスフローを右クリックし、ノードの作成 > オフライン同期 を選択します。
3. ノードの作成 ダイアログボックスで、ノードの名前を入力し、OK をクリックします。
ネットワークとリソースを設定します。
1. データソース セクションで、データソース を SQL Server に設定し、ソースインスタンス名 ドロップダウンリストからデータソースを選択します。
2. リソースグループ セクションで、専用リソースグループを選択します。
3. データ宛先 セクションで、データ宛先 を Elasticsearch に設定し、ソースインスタンス名 ドロップダウンリストからデータソースを選択します。
[次へ] をクリックします。
タスクを設定します。
1. データソース セクションで、同期するテーブルを選択します。
2. データ宛先 セクションで、宛先パラメーターを設定します。
3. フィールドマッピング セクションで、ソースフィールド を ターゲットフィールド にマッピングします。設定の詳細については、「ウィザードモードでの同期タスクの設定」をご参照ください。
4. チャンネル制御 セクションで、チャネルパラメーターを設定します。
パラメーターの詳細については、「ウィザードモードでの同期タスクの設定」をご参照ください。
タスクを実行します。
1. (オプション) タスクのスケジューリングプロパティを設定します。右側のペインで スケジューリング設定 をクリックし、スケジューリングパラメーターを設定します。パラメーターの詳細については、「スケジューリング設定」をご参照ください。
2. ノード設定タブの右上隅にある保存アイコンをクリックして、タスクを保存します。
3. ノード設定タブの右上隅にある送信アイコンをクリックして、タスクを送信します。
  
  タスクにスケジューリングプロパティを設定している場合、タスクは定期的に実行されます。ノード設定タブの右上隅にある実行アイコンをクリックして、タスクをすぐに実行することもできます。
  
  Shell run successfully! を含む実行ログは、タスクが成功したことを示します。

手順 5：データ同期結果の検証

宛先の Alibaba Cloud Elasticsearch インスタンスの Kibana コンソールにログインします。詳細については、「Kibana コンソールへのログイン」をご参照ください。
Kibana ページの左上隅にあるメニューアイコンをクリックし、[開発ツール] を選択します。

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

POST /dbo.students/_search?pretty
{
   "query": { "match_all": {}}
}

説明

dbo.students を、DataWorks データ同期タスクを設定したときに指定した index フィールドの値で置き換えます。

想定される結果は次のとおりです。

{
  "took" : 4,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 3,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "dbo.students",
        "_type" : "_doc",
        "_id" : "XLZFCoUBpTk7S_sKy0dj",
        "_score" : 1.0,
        "_source" : {
          "age" : 21,
          "id" : 1,
          "name" : "Xiaoming"
        }
      },
      {
        "_index" : "dbo.students",
        "_type" : "_doc",
        "_id" : "XbZFCoUBpTk7S_sKy0dj",
        "_score" : 1.0,
        "_source" : {
          "age" : 22,
          "id" : 2,
          "name" : "Xiaozhang"
        }
      },
      {
        "_index" : "dbo.students",
        "_type" : "_doc",
        "_id" : "XrZFCoUBpTk7S_sKy0dj",
        "_score" : 1.0,
        "_source" : {
          "age" : 23,
          "id" : 3,
          "name" : "Xiaowang"
        }
      }
    ]
  }
}

背景

前提条件

課金

手順

手順 1： ソースデータの準備

手順 2： 専用リソースグループの購入と設定

手順 3： データソースの追加

手順 4： オフライン同期タスクの設定と実行

手順 5： データ同期結果の検証

手順 1：ソースデータの準備

手順 2：専用リソースグループの購入と設定

手順 3：データソースの追加

手順 4：オフライン同期タスクの設定と実行

手順 5：データ同期結果の検証