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 は、初期完全データ同期中にソースと宛先の読み取りおよび書き込みリソースを使用します。これにより、ソースと宛先の負荷が増加する可能性があります。ソースまたは宛先のパフォーマンスが不良な場合、ソースまたは宛先のスペックが低い場合、またはデータ量が大きい場合、ソースまたは宛先が使用不可能になる可能性があります。たとえば、DTS は次のケースで大量の読み取りおよび書き込みリソースを占有します。ソースで大量の低速 SQL クエリが実行される場合、1 つ以上のテーブルにプライマリキーがない場合、または宛先でデッドロックが発生する場合などです。この問題を防ぐには、データ同期前にデータ同期がソースと宛先のパフォーマンスに与える影響を評価する必要があります。オフピーク時にデータを同期することを推奨します。たとえば、ソースと宛先の 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 クラスターへのデータ同期」をご参照ください。
ソースと宛先を設定します。ページの下部で、[接続をテストして続行] をクリックします。
データを同期するオブジェクトを設定します。
詳細設定を行います。この例では、デフォルトの詳細設定を使用します。
[データ検証] サブステップで、[すべてのテーブルに _routing ポリシーを適用しない] を選択します。
説明宛先 Elasticsearch クラスターが V7.X の場合、[すべてのテーブルに _routing ポリシーを適用しない] を選択する必要があります。
設定が完了したら、データ同期タスクを保存して事前チェックを実行します。その後、DTS インスタンスを購入してデータ同期タスクを開始します。
DTS インスタンスが購入されると、データ同期タスクが開始されます。[データ同期] ページでタスクのデータ同期の進行状況を確認できます。完全データが同期されると、Elasticsearch クラスターで完全データを確認できます。
ステップ 3 (オプション): 同期結果の検証
Elasticsearch クラスターの Kibana コンソールにログインします。
詳細については、「Kibana コンソールへのログイン」をご参照ください。
Kibana ページの左上隅で、
> [Management] > [開発ツール] を選択し、コンソールで次のコマンドを実行します。完全データの同期結果を検証します。
次のコマンドを実行します。
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" : "_doc", "_id" : "3", "_score" : 1.0, "_source" : { "id" : 3, "name" : "user3", "age" : 43, "hobby" : "game" } }, { "_index" : "es_test", "_type" : "_doc", "_id" : "5", "_score" : 1.0, "_source" : { "id" : 5, "name" : "user5", "age" : 42, "hobby" : "basketball" } }, { "_index" : "es_test", "_type" : "_doc", "_id" : "4", "_score" : 1.0, "_source" : { "id" : 4, "name" : "user4", "age" : 24, "hobby" : "run" } }, { "_index" : "es_test", "_type" : "_doc", "_id" : "2", "_score" : 1.0, "_source" : { "id" : 2, "name" : "user2", "age" : 23, "hobby" : "sport" } }, { "_index" : "es_test", "_type" : "_doc", "_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" : "_doc", "_id" : "3", "_score" : 1.0, "_source" : { "id" : 3, "name" : "user3", "age" : 43, "hobby" : "game" } }, { "_index" : "es_test", "_type" : "_doc", "_id" : "5", "_score" : 1.0, "_source" : { "id" : 5, "name" : "user5", "age" : 42, "hobby" : "basketball" } }, { "_index" : "es_test", "_type" : "_doc", "_id" : "4", "_score" : 1.0, "_source" : { "id" : 4, "name" : "user4", "age" : 24, "hobby" : "run" } }, { "_index" : "es_test", "_type" : "_doc", "_id" : "2", "_score" : 1.0, "_source" : { "id" : 2, "name" : "user2", "age" : 23, "hobby" : "sport" } }, { "_index" : "es_test", "_type" : "_doc", "_id" : "6", "_score" : 1.0, "_source" : { "name" : "user6", "id" : 6, "age" : 30, "hobby" : "dance" } }, { "_index" : "es_test", "_type" : "_doc", "_id" : "1", "_score" : 1.0, "_source" : { "id" : 1, "name" : "user1", "age" : 22, "hobby" : "music" } } ] } }