Elasticsearch:PolarDB for MySQL から Elasticsearch へのデータ同期

仕組み

DTS は PolarDB for MySQL のバイナリログ (binlog) から行レベルの変更をキャプチャし、Elasticsearch にストリーミングします。同期は次の 2 つのフェーズで順次実行されます。

DTS は増分同期にバイナリログに依存するため、タスク開始前に PolarDB クラスターでバイナリロギングを有効にしておく必要があります。DTS は DDL 操作（例：DROP COLUMN や RENAME TABLE）をキャプチャしません。キャプチャされるのは DML 操作（INSERT、DELETE、UPDATE）のみです。この違いは重要です。同期中にソーステーブルに対して DDL 操作を実行すると、Elasticsearch インデックスが同期外れとなり、手動での回復が必要になります。回復手順については、「制限事項」をご参照ください。

前提条件

作業を開始する前に、以下の要件を満たしていることを確認してください。

• バイナリロギングが有効化された PolarDB for MySQL 8.0.1 Enterprise Edition クラスター

• Auto Indexing が有効化された Alibaba Cloud Elasticsearch V7.10 クラスター

• 新しい DTS コンソールへのアクセス権

制限事項

同期の設定と実行

この例では、ソースとして PolarDB for MySQL 8.0.1 Enterprise Edition クラスターを、送信先として Alibaba Cloud Elasticsearch V7.10 クラスターを使用します。

同期結果の確認（任意）

Kibana コンソールを使用して、PolarDB からのデータが正しくレプリケートされたことを確認します。

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

2. 左上隅で > [Management] > [Dev Tools] を選択し、[Console] タブをクリックします。

3. 完全同期の結果を確認します。次のコマンドを実行します。

   GET /product/_search

   応答には、ステップ 1 で挿入した 5 件のレコードすべてが表示されるはずです。

   {
     "took" : 3,
     "timed_out" : false,
     "_shards" : {
       "total" : 5,
       "successful" : 5,
       "skipped" : 0,
       "failed" : 0
     },
     "hits" : {
       "total" : {
         "value" : 5,
         "relation" : "eq"
       },
       "max_score" : 1.0,
       "hits" : [
         {
           "_index" : "product",
           "_type" : "product",
           "_id" : "3",
           "_score" : 1.0,
           "_source" : {
             "id" : 3,
             "name" : "mobile phone C",
             "price" : "2600",
             "code" : "cmp",
             "color" : "black"
           }
         },
         {
           "_index" : "product",
           "_type" : "product",
           "_id" : "5",
           "_score" : 1.0,
           "_source" : {
             "id" : 5,
             "name" : "mobile phone E",
             "price" : "2800",
             "code" : "emp",
             "color" : "silvery"
           }
         },
         {
           "_index" : "product",
           "_type" : "product",
           "_id" : "4",
           "_score" : 1.0,
           "_source" : {
             "id" : 4,
             "name" : "mobile phone D",
             "price" : "2700",
             "code" : "dmp",
             "color" : "red"
           }
         },
         {
           "_index" : "product",
           "_type" : "product",
           "_id" : "2",
           "_score" : 1.0,
           "_source" : {
             "id" : 2,
             "name" : "mobile phone B",
             "price" : "2200",
             "code" : "bmp",
             "color" : "white"
           }
         },
         {
           "_index" : "product",
           "_type" : "product",
           "_id" : "1",
           "_score" : 1.0,
           "_source" : {
             "id" : 1,
             "name" : "mobile phone A",
             "price" : "2000",
             "code" : "amp",
             "color" : "golden"
           }
         }
       ]
     }
   }

4. 増分同期を確認します。ソーステーブルに新しいレコードを挿入します。

   INSERT INTO `test_polardb`.`product` (`id`,`name`,`price`,`code`,`color`) VALUES (6,'mobile phone F','2750','fmp','white');

   レコードがレプリケートされた後、再度 GET /product/_search を実行します。応答には、新しいレコードを含む合計 6 件のヒットが表示されるはずです。

   {
     "took" : 439,
     "timed_out" : false,
     "_shards" : {
       "total" : 5,
       "successful" : 5,
       "skipped" : 0,
       "failed" : 0
     },
     "hits" : {
       "total" : {
         "value" : 6,
         "relation" : "eq"
       },
       "max_score" : 1.0,
       "hits" : [
         {
           "_index" : "product",
           "_type" : "product",
           "_id" : "3",
           "_score" : 1.0,
           "_source" : {
             "id" : 3,
             "name" : "mobile phone C",
             "price" : "2600",
             "code" : "cmp",
             "color" : "black"
           }
         },
         {
           "_index" : "product",
           "_type" : "product",
           "_id" : "5",
           "_score" : 1.0,
           "_source" : {
             "id" : 5,
             "name" : "mobile phone E",
             "price" : "2800",
             "code" : "emp",
             "color" : "silvery"
           }
         },
         {
           "_index" : "product",
           "_type" : "product",
           "_id" : "4",
           "_score" : 1.0,
           "_source" : {
             "id" : 4,
             "name" : "mobile phone D",
             "price" : "2700",
             "code" : "dmp",
             "color" : "red"
           }
         },
         {
           "_index" : "product",
           "_type" : "product",
           "_id" : "2",
           "_score" : 1.0,
           "_source" : {
             "id" : 2,
             "name" : "mobile phone B",
             "price" : "2200",
             "code" : "bmp",
             "color" : "white"
           }
         },
         {
           "_index" : "product",
           "_type" : "product",
           "_id" : "6",
           "_score" : 1.0,
           "_source" : {
             "code" : "fmp",
             "color" : "white",
             "price" : "2750",
             "name" : "mobile phone F",
             "id" : 6
           }
         },
         {
           "_index" : "product",
           "_type" : "product",
           "_id" : "1",
           "_score" : 1.0,
           "_source" : {
             "id" : 1,
             "name" : "mobile phone A",
             "price" : "2000",
             "code" : "amp",
             "color" : "golden"
           }
         }
       ]
     }
   }

次のステップ

• Elasticsearch クラスター内のデータを検索および分析するには、「Alibaba Cloud Elasticsearch とは」をご参照ください。

• 実行中の同期タスクからオブジェクトを削除するには、「データ同期タスクからオブジェクトを削除する」をご参照ください。

• 実行中の同期タスクにオブジェクトを再度追加するには、「データ同期タスクにオブジェクトを追加する」をご参照ください。