MySQL から Alibaba Cloud Elasticsearch へのリアルタイムデータ同期 - Elasticsearch

ApsaraDB RDS for MySQL データベースの本番データを Alibaba Cloud Elasticsearch で検索および分析したい場合、Data Transmission Service (DTS) を使用すると、データを Elasticsearch クラスターにリアルタイムで同期できます。同期はリアルタイムデータ同期タスクに基づいて行われます。このソリューションは、リアルタイムデータ同期で高いパフォーマンスが必要なシナリオに適しています。このトピックでは、ApsaraDB RDS for MySQL データベースから Alibaba Cloud Elasticsearch クラスターへリアルタイムでデータを同期するためのリアルタイムデータ同期タスクの作成方法について説明します。また、全量データと増分データの同期結果を検証する方法についても説明します。

背景情報

DTS は、データ移行、データサブスクリプション、リアルタイムデータ同期を統合したデータ転送サービスです。詳細については、「DTS」をご参照ください。DTS は、挿入、削除、更新操作によって生成されたデータ変更を同期できます。DTS がデータを同期できるデータソースのバージョンについては、「データ同期シナリオの概要」をご参照ください。
DTS を使用して、MySQL から Elasticsearch へ全量データまたは増分データを同期できます。このソリューションは、リレーショナルデータベースからリアルタイムでデータを同期する際に高いパフォーマンスが必要なシナリオや、リレーショナルデータベースから Alibaba Cloud Elasticsearch クラスターへ全量データまたは増分データを同期するシナリオに適しています。

注意事項

DTS は DDL 操作によって生成されたデータ変更を同期しません。データ同期中にソースデータベースのテーブルで DDL 操作が実行された場合、次の操作を実行する必要があります。データ同期タスクからテーブルを削除し、Elasticsearch クラスターからテーブルのインデックスを削除してから、データ同期タスクにテーブルを再度追加してください。詳細については、「同期オブジェクトの削除」および「データ同期タスクへのオブジェクトの追加」をご参照ください。
ソーステーブルに列を追加する場合は、テーブルに対応するインデックスのマッピングを変更してください。次に、ソーステーブルで関連する DDL 操作を実行し、データ同期タスクを一時停止してから、タスクを再度開始してください。
DTS は初期全量データ同期中にソースとデスティネーションの読み取りおよび書き込みリソースを使用します。このため、ソースとデスティネーションの負荷が増加する可能性があります。ソースまたはデスティネーションのパフォーマンスが低い場合、仕様が低い場合、またはデータ量が大きい場合、ソースまたはデスティネーションが利用できなくなる可能性があります。たとえば、ソースで大量のスロー SQL クエリが実行される、1 つ以上のテーブルにプライマリキーがない、またはデスティネーションでデッドロックが発生する場合などに、DTS は大量の読み取りおよび書き込みリソースを占有します。この問題を防ぐために、データ同期前にソースとデスティネーションのパフォーマンスへのデータ同期の影響を評価する必要があります。オフピーク時にデータを同期することを推奨します。たとえば、ソースとデスティネーションの CPU 使用率が 30% 未満の場合にデータを同期できます。
- ピーク時に全量データを同期すると、同期が失敗する可能性があります。この場合、同期タスクを再起動してください。
- ピーク時に増分データを同期すると、データ同期のレイテンシーが発生する可能性があります。
ApsaraDB RDS for MySQL と Elasticsearch は異なるデータ型をサポートしています。初期スキーマ同期中、DTS はデスティネーションでサポートされるデータ型に基づいて、ソースフィールドとデスティネーションフィールド間のマッピングを作成します。詳細については、「スキーマ同期のデータ型マッピング」をご参照ください。

プロセス

準備：同期するデータをソースの ApsaraDB RDS for MySQL データベースに追加し、Alibaba Cloud Elasticsearch クラスターを作成し、Elasticsearch クラスターの自動インデックス作成機能を有効にします。
データ同期タスクの作成と実行：DTS コンソールでデータ同期タスクを作成して実行します。
データ同期結果の検証：Elasticsearch クラスターの Kibana コンソールにログインして、全量データの同期結果を検証します。次に、ソースの ApsaraDB RDS for MySQL データベースにデータを追加し、Elasticsearch クラスターの Kibana コンソールにログインして、増分データの同期結果を検証します。

操作手順

ステップ 1: 準備

この例では、MySQL 8.0 の ApsaraDB RDS インスタンスと Alibaba Cloud Elasticsearch V7.10 クラスターを準備します。

ソースデータベースと同期するデータの準備

MySQL 8.0 の ApsaraDB RDS インスタンスを作成します。詳細については、「ApsaraDB RDS for MySQL インスタンスの作成」をご参照ください。
データベースアカウントと test_mysql という名前のデータベースを作成します。詳細については、「データベースとアカウントの作成」をご参照ください。

test_mysql データベースで、es_test テーブルを作成してデータを挿入します。テーブル作成ステートメントとデータは次のとおりです：

-- テーブル作成
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');

デスティネーション Elasticsearch クラスターの準備

Alibaba Cloud Elasticsearch V7.10 クラスターを作成します。詳細については、「Alibaba Cloud Elasticsearch クラスターの作成」をご参照ください。
Elasticsearch クラスターの自動インデックス作成機能を有効にします。詳細については、「YML ファイルの設定」をご参照ください。

ステップ 2: データ同期タスクの作成と実行

新しい DTS コンソールの [データ同期] ページに移動します。
Create Task をクリックします。
表示されるページで、プロンプトに従ってデータ同期タスクを作成および設定します。

説明
次の手順に含まれるパラメーターについては、「ApsaraDB RDS for MySQL インスタンスから Elasticsearch クラスターへのデータ同期」をご参照ください。
1. ソースとデスティネーションを設定します。ページの下部で、[接続をテストして続行] をクリックします。
2. 同期するオブジェクトを設定します。
3. 詳細設定を設定します。この例では、デフォルトの詳細設定を使用します。
4. [データ検証] サブステップで、[_routing ポリシーをどのテーブルにも適用しない] を選択します。
  
  説明
  デスティネーションの Elasticsearch クラスターが V7.X の場合、[すべてのテーブルに _routing ポリシーを適用しない] を選択する必要があります。
設定が完了したら、データ同期タスクを保存し、タスクの事前チェックを実行し、DTS インスタンスを購入してデータ同期タスクを開始します。

DTS インスタンスが購入されると、データ同期タスクの実行が開始されます。[データ同期] ページでタスクのデータ同期の進行状況を確認できます。全量データの同期後、Elasticsearch クラスターで全量データを確認できます。

ステップ 3 (オプション): 同期結果の検証

Elasticsearch クラスターの Kibana コンソールにログインします。

詳細については、「Kibana コンソールへのログイン」をご参照ください。
Kibana ページの左上隅で、 > [管理] > [開発ツール] を選択し、コンソールで次のコマンドを実行します。

全量データの同期結果を検証します。

次のコマンドを実行します：

GET /es_test/_search

コマンドが正常に実行されると、次の結果が返されます：

{
  "took" : 10,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 5,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "3",
        "_score" : 1.0,
        "_source" : {
          "id" : 3,
          "name" : "user3",
          "age" : 43,
          "hobby" : "game"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "5",
        "_score" : 1.0,
        "_source" : {
          "id" : 5,
          "name" : "user5",
          "age" : 42,
          "hobby" : "basketball"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "4",
        "_score" : 1.0,
        "_source" : {
          "id" : 4,
          "name" : "user4",
          "age" : 24,
          "hobby" : "run"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "2",
        "_score" : 1.0,
        "_source" : {
          "id" : 2,
          "name" : "user2",
          "age" : 23,
          "hobby" : "sport"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "id" : 1,
          "name" : "user1",
          "age" : 22,
          "hobby" : "music"
        }
      }
    ]
  }
}

増分データの同期結果を検証します。

次のステートメントを実行して、ソーステーブルにデータレコードを挿入します：
```
INSERT INTO `test_mysql`.`es_test` (`id`,`name`,`age`,`hobby`) VALUES (6,'user6',30,'dance');
```

増分データの同期が完了したら、コマンド GET /es_test/_search を再度実行して結果を確認します。

コマンドが正常に実行されると、次の結果が返されます：

{
  "took" : 541,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 6,
      "relation" : "eq"
    },
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "3",
        "_score" : 1.0,
        "_source" : {
          "id" : 3,
          "name" : "user3",
          "age" : 43,
          "hobby" : "game"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "5",
        "_score" : 1.0,
        "_source" : {
          "id" : 5,
          "name" : "user5",
          "age" : 42,
          "hobby" : "basketball"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "4",
        "_score" : 1.0,
        "_source" : {
          "id" : 4,
          "name" : "user4",
          "age" : 24,
          "hobby" : "run"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "2",
        "_score" : 1.0,
        "_source" : {
          "id" : 2,
          "name" : "user2",
          "age" : 23,
          "hobby" : "sport"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "6",
        "_score" : 1.0,
        "_source" : {
          "name" : "user6",
          "id" : 6,
          "age" : 30,
          "hobby" : "dance"
        }
      },
      {
        "_index" : "es_test",
        "_type" : "es_test",
        "_id" : "1",
        "_score" : 1.0,
        "_source" : {
          "id" : 1,
          "name" : "user1",
          "age" : 22,
          "hobby" : "music"
        }
      }
    ]
  }
}