注意事項

• DTS在執行全量數據初始化時將占用源庫和目標庫一定的讀寫資源，可能會導致數據庫的負載上升，在數據庫性能較差、規格較低或業務量較大的情況下（例如源庫有大量慢SQL、存在無主鍵表或目標庫存在死鎖等），可能會加重數據庫壓力，甚至導致數據庫服務不可用。因此您需要在執行數據同步前評估源庫和目標庫的性能，同時建議您在業務低峰期執行數據同步（例如源庫和目標庫的CPU負載在30%以下）。

• 如果源庫中待同步的表需要執行增加列的操作，您只需先在Elasticsearch實例中修改對應表的mapping，然后在源庫中執行相應的DDL操作，最后暫停并啟動DTS同步實例即可。

使用限制

• 通過DTS將數據同步至阿里云Elasticsearch，不支持7.16和8.x版本的Elasticsearch實例。

• DTS不支持同步DDL操作，如果源庫中待同步的表在同步的過程中已經執行了DDL操作，您需要先移除同步對象，然后在Elasticsearch實例中移除該表對應的索引，最后新增同步對象。詳情請參見移除同步對象和新增同步對象。

操作流程

1. 步驟一：環境準備

   完成創建Elasticsearch實例、創建PolarDB MySQL集群、準備測試數據等任務。

2. 步驟二：配置數據同步鏈路

   通過數據傳輸服務DTS，快速創建并啟動PolarDB MySQL到阿里云Elasticsearch的實時同步作業。

3. 步驟三：查看數據同步結果

   在阿里云Elasticsearch的Kibana控制臺中查看同步成功的數據。

4. 步驟四：驗證增量數據同步

   驗證在PolarDB MySQL數據庫中新增數據時，數據的同步效果。

步驟一：環境準備

1. 在阿里云Elasticsearch實例中開啟自動創建索引功能。

   具體操作步驟請參見配置YML參數。本文使用6.7版本的實例。

   說明

   阿里云Elasticsearch為了保證用戶操作數據的安全性，默認把自動創建索引的配置設置為不允許。通過DTS同步數據，使用的是提交數據的方式創建索引，而不是Create index API方式。所以在同步數據前，需要先開啟集群的自動創建索引功能。

2. 創建云數據庫PolarDB MySQL集群，并開啟Binlog。

   具體操作步驟請參見購買按量付費集群、開啟Binlog。

3. 創建PolarDB MySQL數據庫和表，并插入測試數據。

   具體操作步驟請參見管理數據庫。本文使用的表結構和測試數據如下。

   • 建表語句

     CREATE TABLE `product` (
         `id` bigint(32) NOT NULL AUTO_INCREMENT,
         `name` varchar(32) NULL,
         `price` varchar(32) NULL,
         `code` varchar(32) NULL,
         `color` varchar(32) NULL,
         PRIMARY KEY (`id`)
     ) ENGINE=InnoDB
     DEFAULT CHARACTER SET=utf8;

   • 測試數據

     INSERT INTO `estest`.`product` (`id`,`name`,`price`,`code`,`color`) VALUES (1,'mobile phone A','2000','amp','golden');
     INSERT INTO `estest`.`product` (`id`,`name`,`price`,`code`,`color`) VALUES (2,'mobile phone B','2200','bmp','white');
     INSERT INTO `estest`.`product` (`id`,`name`,`price`,`code`,`color`) VALUES (3,'mobile phone C','2600','cmp','black');
     INSERT INTO `estest`.`product` (`id`,`name`,`price`,`code`,`color`) VALUES (4,'mobile phone D','2700','dmp','red');
     INSERT INTO `estest`.`product` (`id`,`name`,`price`,`code`,`color`) VALUES (5,'mobile phone E','2800','emp','silvery');

步驟二：配置數據同步鏈路

1. 進入同步任務的列表頁面。

   1. 登錄DMS數據管理服務。

   2. 在頂部菜單欄中，單擊集成與開發（DTS）。

   3. 在左側導航欄，選擇數據傳輸（DTS） > 數據同步。

   說明

   • 實際操作可能會因DMS的模式和布局不同，而有所差異。更多信息，請參見極簡模式和自定義DMS界面布局與樣式。

   • 您也可以登錄新版DTS同步任務的列表頁面。

2. 單擊創建任務，按照頁面提示創建并配置數據同步任務。

   您需要依次完成源庫及目標庫配置、任務對象配置、映射字段配置、高級配置和庫表字段配置，本文使用的配置及相關說明如下，更多詳細信息請參見PolarDB MySQL為源的數據同步和PolarDB-X 2.0同步至Elasticsearch。

   1. 配置源庫及目標庫。

      

      類別	配置	說明
      無	任務名稱	DTS為每個任務自動生成一個同步作業名稱，該名稱沒有唯一性要求。 建議配置具有業務意義的名稱，便于后續的識別。
      源庫信息	數據庫類型	選擇PolarDB MySQL。
      	接入方式	選擇云實例。
      	實例地區	選擇源PolarDB MySQL數據庫所屬地域。
      	是否跨阿里云賬號	本場景為同一阿里云賬號間同步數據，選擇不跨賬號。
      	實例ID	選擇源PolarDB MySQL實例ID。
      	數據庫賬號	填入源PolarDB MySQL實例的數據庫賬號，需具備待同步對象的讀權限。
      	數據庫密碼	填入數據庫賬號對應的密碼。
      目標庫信息	數據庫類型	選擇ElasticSearch。
      	接入方式	固定為云實例。
      	實例地區	選擇目標Elasticsearch實例所屬地域，建議與源MySQL數據庫保持一致。
      	實例ID	選擇目標Elasticsearch實例ID。
      	數據庫賬號	填入連接Elasticsearch實例的賬號，默認賬號為elastic。
      	數據庫密碼	填入數據庫賬號對應的密碼。

   2. 配置任務對象。

      配置	說明
      任務步驟	固定選中增量同步。默認情況下，您還需要同時選中庫表結構同步和全量同步。預檢查完成后，DTS會將源實例中待同步對象的全量數據在目標集群中初始化，作為后續增量同步數據的基線數據。
      目標已存在表的處理模式	預檢查并報錯攔截：檢查目標數據庫中是否有同名的表。如果目標數據庫中沒有同名的表，則通過該檢查項目；如果目標數據庫中有同名的表，則在預檢查階段提示錯誤，數據同步任務不會被啟動。 說明 如果目標庫中同名的表不方便刪除或重命名，您可以更改該表在目標庫中的名稱，請參見庫表列名映射。 忽略報錯并繼續執行：跳過目標數據庫中是否有同名表的檢查項。 警告 選擇為忽略報錯并繼續執行，可能導致數據不一致，給業務帶來風險，例如： 表結構一致的情況下，如在目標庫遇到與源庫主鍵或唯一鍵的值相同的記錄： 全量期間，DTS會保留目標集群中的該條記錄，即源庫中的該條記錄不會同步至目標數據庫中。 增量期間，DTS不會保留目標集群中的該條記錄，即源庫中的該條記錄會覆蓋至目標數據庫中。 表結構不一致的情況下，可能會導致無法初始化數據、只能同步部分列的數據或同步失敗，請謹慎操作。
      索引名稱	表名 選擇為表名后，在目標Elasticsearch實例中創建的索引名稱和表名一致，在本案例中即為product。 庫名_表名 選擇為庫名_表名后，在目標Elasticsearch實例中創建的索引名稱為庫名_表名，在本案例中即為estest_product。
      同步對象	在源庫對象框中單擊待同步對象，然后單擊將其移動至已選擇對象框。
      映射名稱更改	如需更改單個同步對象在目標實例中的名稱，請右擊已選擇對象中的同步對象，設置方式，請參見庫表列名單個映射。 如需批量更改同步對象在目標實例中的名稱，請單擊已選擇對象方框右上方的批量編輯，設置方式，請參見庫表列名批量映射。

   3. 配置映射字段，修改同步后的字段名稱。

      如果您需要修改同步后的字段名稱，可在已選擇對象區域框中，右鍵單擊對應的表名，設置該表在目標Elasticsearch實例中的索引名稱、Type名稱等信息，然后單擊確定。本文使用的配置及相關說明如下，未提及的配置保持默認，更多詳細信息請參見庫表列名單個映射。

      配置	說明
      索引名稱	自定義索引名稱，詳情請參見基本概念。 重要 輸入索引名稱時，請確保Elasticsearch集群中不存在同名索引，否則報錯index already exists。
      Type名稱	自定義索引類型名稱，詳情請參見基本概念。
      過濾條件	您可以設置SQL過濾條件，過濾待同步的數據，只有滿足過濾條件的數據才會被同步到目標實例，詳情請參見通過SQL條件過濾任務數據。
      字段參數	選擇所需的字段參數和字段參數值，字段參數及取值介紹請參見Mapping parameters。 重要 如果添加參數時，將對應參數的index值設置為false，那么該字段將不能被查詢，詳情請參見index。

   4. 配置高級參數。

      本文使用默認配置，相關說明如下。

      配置	說明
      選擇調度該任務的專屬集群	DTS默認調度到共享集群上，用戶不需要選擇調度到指定的集群節點上。 專屬集群需要先購買后，運行成功才能選擇使用 您可以購買指定規格的專屬集群來運行DTS遷移、同步、訂閱等任務，實現與其他用戶的DTS任務資源隔離，專屬集群中的DTS實例相比共享集群的實例，實現資源獨享，穩定性更好，性能更優。
      監控報警	是否設置告警，當同步失敗或延遲超過閾值后，將通知告警聯系人。 不設置：不設置告警。 設置：設置告警，您還需要設置告警閾值和告警聯系人。更多信息，請參見在配置任務過程中配置監控報警。
      源庫、目標庫無法連接后的重試時間	在同步任務啟動后，若源庫或目標庫連接失敗則DTS會報錯，并會立即進行持續的重試連接，默認持續重試時間為720分鐘，您也可以在取值范圍（10~1440分鐘）內自定義重試時間，建議設置30分鐘以上。如果DTS在設置的重試時間內重新連接上源庫、目標庫，同步任務將自動恢復。否則，同步任務將會失敗。 說明 針對同源或者同目標的多個DTS實例，如DTS實例A和DTS實例B，設置網絡重試時間時A設置30分鐘，B設置60分鐘，則重試時間以低的30分鐘為準。 由于連接重試期間，DTS將收取任務運行費用，建議您根據業務需要自定義重試時間，或者在源和目標庫實例釋放后盡快釋放DTS實例。
      分片配置	根據目標Elasticsearch中索引的分片配置，設置索引的主分片和副本分片的數量。Elasticsearch 7.x以下版本的索引默認包含5個主shard，1個副shard。Elasticsearch 7.x及以上版本的索引默認包含1個主shard，1個副shard。 重要 shard大小和數量是影響Elasticsearch集群穩定性和性能的重要因素，您需要設置合理的shard數，shard的評估方法請參見規格容量評估。
      字符串Index	同步至目標Elasticsearch實例中的字符串編入索引的方式： analyzed：先分析字符串，再寫入索引。您還需要選擇具體的分析器，分析器的類型及作用，請參見分析器。 not analyzed：不分析，直接使用原始值寫入索引。 no：不寫入索引。
      時區	DTS同步時間類型的數據（如DATETIME、TIMESTAMP）至目標Elasticsearch實例時，您可以選擇所帶時區。 說明 如目標實例中此類時間類型數據無需帶有時區，則在同步前您需在目標實例中設置該時間類型數據的文檔類型（type）。
      DOCID取值	DOCID默認為表的主鍵，如表中無主鍵，則DOCID為Elasticsearch自動生成的ID列。
      配置ETL功能	ETL配置功能暫不支持在目標端進行表結構變更，涉及變更操作需要您在鏈路啟動前預先在目標端完成。 ETL配置的二次修改可能因您的配置造成鏈路中斷或歷史數據修改，請謹慎操作。 ETL配置二次修改啟動時僅針對啟動后的增量數據進行適配，歷史數據不支持修改后的配置。 詳情參見：在DTS遷移或同步任務中配置ETL

   5. 配置庫表字段，設置待同步的表在目標Elasticsearch的_routing策略和_id取值。

      本文使用的數據庫名稱為estest，表名稱為product，設置_routing為否，_id取值為id，相關說明如下。

      類型	說明
      設置_routing	設置_routing可以將文檔路由存儲在目標Elasticsearch實例的指定分片上，請參見_routing。 選擇為是，您可以自定義列進行路由。 選擇為否，則用_id進行路由。 說明 創建的目標Elasticsearch實例為7.x版本時，您必須選擇為否。
      _id取值	表的主鍵列 聯合主鍵合并為一列。 業務主鍵 如果選擇為業務主鍵，那么您還需要設置對應的業務主鍵列。

3. 配置完成后，根據頁面提示保存任務、進行預檢查、購買并啟動任務。

   購買成功后，同步任務正式開始，您可在數據同步界面查看具體任務進度。待全量同步完成，增量同步進行中時，您即可在Elasticsearch中查看同步成功的數據。

步驟三：查看數據同步結果

1. 登錄目標阿里云Elasticsearch實例的Kibana控制臺，根據頁面提示進入Kibana主頁。
   登錄Kibana控制臺的具體操作，請參見登錄Kibana控制臺。

   說明 本文以阿里云Elasticsearch 6.7.0版本為例，其他版本操作可能略有差別，請以實際界面為準。
2. 在左側導航欄，單擊Dev Tools。

3. 在Console中，執行如下命令查看全量數據同步結果。

   GET /product/_doc/_search

   預期結果如下。

4. 通過控制臺操作查看同步成功的數據。

   1. 為目標索引創建索引模式。

      1. 在左側導航欄，單擊Management。

      2. 在Kibana區域，單擊Index Patterns。

      3. 單擊Create index pattern。

      4. 在Create index pattern頁面的Index pattern name中，輸入與同步成功的索引相匹配的索引模式名稱（例如product、product*）。

      5. 單擊Next step。

      6. 單擊Create index pattern。

   2. 在左側導航欄，單擊Discover。

   3. 選擇您已創建的索引模式，查看同步成功的數據。

      

步驟四：驗證增量數據同步

1. 進入PolarDB控制臺。

2. 在PolarDB MySQL數據庫中插入一條數據。

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

3. 登錄Kibana控制臺，在左側導航欄，單擊Discover，選擇您已創建的索引模式，查看同步成功的增量數據。

   

   說明

   您也可以使用同樣的方式，在PolarDB MySQL數據庫中刪除或修改數據，然后查看數據同步結果。