騰訊云數(shù)據(jù)庫DTS發(fā)布全新數(shù)據(jù)集成方案：全增量無縫同步，快速構(gòu)建實時數(shù)倉

隨著IT技術(shù)與大數(shù)據(jù)的不斷發(fā)展，越來越多的企業(yè)開始意識到數(shù)據(jù)的價值，通過大數(shù)據(jù)分析，可以幫助企業(yè)更深入地了解用戶需求、更好地洞察市場趨勢。目前大數(shù)據(jù)分析在每個業(yè)務(wù)運營中都發(fā)揮著重要作用，成為企業(yè)提升市場競爭力的關(guān)鍵舉措之一。通常企業(yè)會構(gòu)建數(shù)據(jù)湖倉，將多個數(shù)據(jù)源通過數(shù)據(jù)集成技術(shù)，匯集一起進行數(shù)據(jù)分析。由此，數(shù)據(jù)集成成為了構(gòu)建數(shù)據(jù)湖倉的必經(jīng)之路，然而企業(yè)在數(shù)據(jù)集成過程中卻面臨很多棘手問題。

·全量+增量數(shù)據(jù)集成割裂

傳統(tǒng)的數(shù)據(jù)集成大多僅支持全量數(shù)據(jù)，對于全量+增量的一并集成，則需要分別部署鏈路，獲取到數(shù)據(jù)后再手動合并。

·多個數(shù)據(jù)源頭，操作與維護復雜

表結(jié)構(gòu)頻繁變更，無法自動同步表結(jié)構(gòu)變更到數(shù)據(jù)湖倉，手動維護成本高。另外無法”一鍵”整庫同步，追加同步對象操作復雜等。

·數(shù)據(jù)獲取時效性差

傳統(tǒng)的數(shù)據(jù)集成技術(shù)建模路徑較長，按照T+1的方式同步到數(shù)據(jù)倉庫中，時效性差。需要做到實時數(shù)據(jù)集成和分析，才能幫助用戶根據(jù)最新的數(shù)據(jù)做出更快、更準確的決策。

基于數(shù)據(jù)集成的核心痛點和用戶訴求，近期騰訊云數(shù)據(jù)傳輸服務(wù)DTS重磅發(fā)布全新數(shù)據(jù)集成方案，該方案采取全增量數(shù)據(jù)一起的同步方式，將數(shù)據(jù)源先同步到Ckafka，再從Ckafka消費數(shù)據(jù)投遞到數(shù)據(jù)湖倉，可以有效幫助用戶解決數(shù)據(jù)湖倉建設(shè)前期數(shù)據(jù)集成的問題。

關(guān)于DTS

選擇DTS做數(shù)據(jù)集成是因為DTS有著技術(shù)上的天然優(yōu)勢。

2.1 DTS簡介

DTS是騰訊云自主研發(fā)的專注于數(shù)據(jù)庫傳輸服務(wù)的工具，具有高傳輸性能、高可用、安全連接、操作便捷等特點，可以實現(xiàn)數(shù)據(jù)源在業(yè)務(wù)不停服狀態(tài)下的實時數(shù)據(jù)同步，整個數(shù)據(jù)同步過程對源庫業(yè)務(wù)無影響。DTS已成功應(yīng)用于金融、醫(yī)療、娛樂、泛互聯(lián)網(wǎng)等多個行業(yè)場景，幫助用戶實現(xiàn)不同系統(tǒng)的數(shù)據(jù)打通和自由流動，如數(shù)據(jù)庫遷移上云、數(shù)據(jù)庫異地備份、異地多活等。

2.2 DTS的技術(shù)優(yōu)勢

首先，DTS本身已支持多種數(shù)據(jù)源的同步，涵蓋MySQL、MariaDB、Percona、TDSQL-C MySQL版、TDSQL MySQL版、PostgreSQL、Redis、MongoDB、SQL Server等，對各種類型的數(shù)據(jù)庫以及對應(yīng)的數(shù)據(jù)格式都“了如指掌”，可以保證數(shù)據(jù)同步結(jié)果的正確性。

在DTS已有的技術(shù)積累中，已支持了無鎖同步技術(shù)，即在同步過程中，不會對源庫加全局只讀鎖（FTWRL），避免影響源庫的寫入。在數(shù)據(jù)同步過程中，源庫若發(fā)生主從切換、重啟等，任務(wù)都可以正常運行，不會被中斷。這些技術(shù)都對源庫非常友好，保證使用DTS同步數(shù)據(jù)的同時，不影響源庫業(yè)務(wù)的正常運行。

其次，提供全增量一體的數(shù)據(jù)集成能力是當前業(yè)界的主流發(fā)展方向，而DTS本身就具備此能力，DTS在數(shù)據(jù)庫之間的同步機制，原生就采用全增量無縫銜接的同步機制，既能保證數(shù)據(jù)一致性，又能保證數(shù)據(jù)的實時性。這個能力可以避免因使用不同工具分別集成全量或增量導致的難以保證數(shù)據(jù)連貫性和一致性的問題。

最后，基于DTS的操作和維護都是Web界面，用戶只需簡單的3-4步即可完成配置，非常便利。

基于DTS本身具有的技術(shù)優(yōu)勢，且技術(shù)日漸成熟和完善，加上用戶對大數(shù)據(jù)集成訴求的日益突出，DTS構(gòu)建數(shù)據(jù)集成的能力也就應(yīng)時而生。

2.3基于DTS的數(shù)據(jù)集成方案

DTS在做數(shù)據(jù)集成方案的初期，產(chǎn)研團隊做了非常充分的調(diào)研，并分析出了用戶的核心訴求，主要聚焦以下四個方面：

支持全量+增量數(shù)據(jù)同步：方便快速將全量+增量數(shù)據(jù)全部同步至下游數(shù)據(jù)分析工具中。

按序消費：數(shù)據(jù)消費時需要按照數(shù)據(jù)生產(chǎn)的順序進行。

數(shù)據(jù)不丟失：數(shù)據(jù)在下游消費時至少要出現(xiàn)一次，不能丟失業(yè)務(wù)數(shù)據(jù)。

維護便捷：庫表結(jié)構(gòu)變更，或者庫表對象追加需要方便操作。

DTS的「數(shù)據(jù)訂閱」模塊可以應(yīng)用于數(shù)據(jù)集成并分發(fā)到下游的場景中，但訂閱模塊主要處理增量數(shù)據(jù)，無法實現(xiàn)全量+增量一起同步。經(jīng)過多次的技術(shù)探討和驗證后，我們最終決定基于「數(shù)據(jù)同步」模塊來做數(shù)據(jù)集成，技術(shù)方案：數(shù)據(jù)源先通過DTS同步數(shù)據(jù)到Ckafka，再從Ckafka消費數(shù)據(jù)投遞到數(shù)據(jù)湖倉。

不過實際落地中，我們還是遇到了一些挑戰(zhàn)。

2.3.1全量部分數(shù)據(jù)塊很大，如何提升導出導入效率？

使用DTS數(shù)據(jù)同步模塊來做數(shù)據(jù)集成，可以滿足全量+增量一起同步的訴求，但在大數(shù)據(jù)場景下，又不得不面臨兩個問題：對于大表（如10億行以上），如何提升同步作業(yè)效率？對于超大的存量數(shù)據(jù)，在全量階段遇到任務(wù)中斷時，如何確保數(shù)據(jù)重入？

基于以上問題，DTS設(shè)計了分塊導出方案，針對大表場景（如10億行以上），從源庫導出數(shù)據(jù)時將一張大表分為多個分塊，一個分塊連接一個線程，這樣一張大表就可實現(xiàn)多分塊同時導出，提升大表的同步效率。

在導入到目標kafka時，也是按照分塊導入的，同時這些分塊都會進行標記，如果kafka發(fā)生重啟，可以根據(jù)標記來識別中斷的分塊位置，從中斷的分塊開始繼續(xù)向目標kafka寫入。使用這個方式，在遇到kafka異常時，就不需要從頭重新寫，大大提升用戶體驗。

2.3.2多分區(qū)，如何保證按序消費？

為了提升用戶消費的速率，消息投遞到Kafka時一般采用投遞到kafka的多個分區(qū)的形式，多個分區(qū)可以并行消費以提升消費速率，但在多分區(qū)處理過程中，會涉及投遞順序的問題，需要保證投遞到每個分區(qū)的消息與業(yè)務(wù)生產(chǎn)的消息順序保持一致。

在實現(xiàn)中，DTS向Kafka投遞消息時，按照源庫日志解析后的順序來寫入，因此可以實現(xiàn)寫入Kafka順序與業(yè)務(wù)生成順序的一致。

·全局順序性

DTS在拉取源庫的binlog日志時，采用單線程機制，先保證日志解析結(jié)果與業(yè)務(wù)生產(chǎn)順序保持一致，等寫入到kafka的多個分區(qū)時，再按照多線程并發(fā)，最終實現(xiàn)了每個分區(qū)的消息都是按序排列。

這里需要說明下，投遞到多Topic+多分區(qū)這種形式中，每個分區(qū)內(nèi)的消息都是按順序投遞的，但是多個分區(qū)同時消費時，無法保證分區(qū)間按序消費，如果用戶對消費到的消息順序有嚴格要求，建議選擇投遞到單Topic+單分區(qū)的形式。

·表級別順序性

在選擇按表名分區(qū)的場景中，源庫同一個表的數(shù)據(jù)變更都會投遞到目標Topic下的同一個分區(qū)中，因為日志的解析是按序排列，所以投遞到Topic分區(qū)中的消息也是按序排列。

總之，不論選擇哪種分區(qū)策略，DTS都可以保證投遞到各分區(qū)中消息的順序性。

2.3.3如何保證數(shù)據(jù)不丟？

要保證同步到Kafka的數(shù)據(jù)一條都不丟，那么所有的數(shù)據(jù)就需要有跡可循，哪些已經(jīng)同步過了、哪些還沒有同步過，都必須清楚可查。于是DTS通過對數(shù)據(jù)做標記，標識數(shù)據(jù)同步位置，以此來實現(xiàn)數(shù)據(jù)準確同步。

全量階段，數(shù)據(jù)按照分塊機制進行導出導入，DTS導入到目標端Kafka的每個分塊都會進行標記，kafka異常時，可以識別中斷的分塊位置繼續(xù)導入。

增量階段，DTS內(nèi)部處理源庫的日志解析時會插入標記，來識別數(shù)據(jù)寫入到Kafka的位置，如果任務(wù)中斷再恢復，通過DTS內(nèi)部標記，可以找到中斷的位置，繼續(xù)增量同步。

2.3.4庫表變更，能否靈活同步？

業(yè)務(wù)數(shù)據(jù)庫經(jīng)常會有庫表結(jié)構(gòu)的變更，而數(shù)據(jù)集成需要能識別并自動同步這些變更字段，否則，庫表結(jié)構(gòu)每變更一次，就需要手動改一次集成程序，這個維護工作量非常大。在DTS以前的鏈路傳輸中，庫表結(jié)構(gòu)變更的自動同步能力就已經(jīng)具備了，直接集成即可。但是我們本次需要解決的是，當同步任務(wù)已經(jīng)啟動，用戶想要追加/刪除一個新的庫表對象，如何做到一鍵化操作，讓用戶便捷維護。

這里，我們以追加一個表對象為例，同步任務(wù)已經(jīng)在進行中，但是運行過程中發(fā)現(xiàn)需要新增一個表對象（例如表A），對用戶來說，只需要在DTS任務(wù)列表頁，進行一步可視化點擊操作即可完成。

動態(tài)修改同步對象的過程中，其實DTS底層做了很多工作，對用戶操作層面進行了簡化，如上述操作案例：新增一個表對象（例如表A），DTS需要同步表A的歷史存量數(shù)據(jù)，同時，已有的同步任務(wù)1還不能受影響。所以在實現(xiàn)中，我們在DTS后臺構(gòu)造了一個臨時任務(wù)2，來負責同步表A的存量數(shù)據(jù)，當任務(wù)2完成后，再將任務(wù)1和任務(wù)2合并，以此來實現(xiàn)動態(tài)追加同步對象的效果。

相對于一般的集成工具，DTS在庫表結(jié)構(gòu)的變更，庫表對象增加/刪除等方面都是非常友好的，用戶只需要在Web界面進行操作，一次配置，即可享受長期便利，大大減少用戶的維護成本。

接下來，給大家重點介紹DTS的數(shù)據(jù)集成方案是如何配置的。

DTS+Ckafka+數(shù)據(jù)湖倉生產(chǎn)實踐

3.1實踐場景

數(shù)據(jù)源頭為MySQL，通過DTS獲取MySQL的全量+增量數(shù)據(jù)到消息隊列Ckafka，然后適配消費Demo，將消息投遞到數(shù)據(jù)湖倉。

3.2前期準備

·準備騰訊云Ckafka實例，并創(chuàng)建好消費組和消費topic。

·準備源數(shù)據(jù)庫MySQL。

·準備執(zhí)行DTS任務(wù)的賬號，并授權(quán)源庫和目標庫的對應(yīng)權(quán)限。

·準備數(shù)據(jù)湖倉。

3.3數(shù)據(jù)同步

DTS的操作比較簡單，在騰訊云Web界面進行4個步驟即可，無需環(huán)境部署。

步驟1：創(chuàng)建DTS任務(wù)

購買一個DTS任務(wù)，源庫選擇MySQL，目標庫選擇Ckafka。

步驟2：設(shè)置同步源和目標數(shù)據(jù)庫

配置DTS連接源庫和目標庫，源庫配置中填入MySQL的主機地址/端口/用戶名/密碼，目標庫選擇Ckafka實例ID。

這個步驟主要是驗證DTS到源和目標庫的網(wǎng)絡(luò)是否打通，對應(yīng)的用戶權(quán)限是否滿足要求，如果源庫有安全組設(shè)置需要允許DTS IP訪問，否則網(wǎng)絡(luò)不通。

步驟3：配置數(shù)據(jù)同步選項

這個步驟主要是選擇同步的數(shù)據(jù)格式（Avro、JSON）、數(shù)據(jù)投遞到Ckafka的哪個topic下、分區(qū)策略等。

·對于庫表結(jié)構(gòu)的變更，一鍵勾選DDL，即可在后續(xù)自動同步庫表結(jié)構(gòu)的變更數(shù)據(jù)。

·選定同步的庫表對象后，如果有需要追加，在任務(wù)啟動后通過修改任務(wù)即可添加。

步驟4：校驗任務(wù)

上述配置完成后，DTS會對源和目標庫的各項參數(shù)進行預校驗，如Binlog必須開啟，并且binlog_format需要設(shè)置為row模式等等，以保證數(shù)據(jù)同步結(jié)果的正確性。預校驗通過后同步任務(wù)就可以啟動了。

3.4數(shù)據(jù)消費和投遞

步驟1：下載消費Demo樣例

DTS同步任務(wù)正常運行后，下載DTS消費Demo樣例，將Demo包解壓后運行，進行數(shù)據(jù)消費。

這里以Go語言為例，解壓Demo包后運行g(shù)o build-o subscribe./main/main.go，生成可執(zhí)行文件subscribe。

然后運行./subscribe--brokers=xxx--topic=xxx--group=xxx--trans2sql=true。這里的brokers、topic、group分別填入Ckafka的地址、消費topic名稱、消費組名稱。

運行結(jié)果顯示如下，表示Kafka正常連接，消費鏈路已打通。

步驟2：測試數(shù)據(jù)結(jié)果

在源數(shù)據(jù)庫上插入一條數(shù)據(jù)。

在消費端即可查看到對應(yīng)數(shù)據(jù)。

步驟3：修改Demo，增加適配到后端數(shù)倉的代碼邏輯

DTS提供的消費Demo僅對數(shù)據(jù)做了打印處理，用戶需要在Demo基礎(chǔ)上自行編寫數(shù)據(jù)處理到后端數(shù)據(jù)湖倉的適配邏輯。

實踐效果

使用DTS同步到Kafka的鏈路形式替代之前使用Canal組件的鏈路，最終實現(xiàn)高性能傳輸、高穩(wěn)定性保障的同時有效降低了運維成本。

傳輸性能高：DTS的傳輸性能與用戶實際網(wǎng)絡(luò)延時、帶寬、數(shù)據(jù)本身的規(guī)格配置都有關(guān)系，在用戶源端和目標端規(guī)格都比較高，網(wǎng)絡(luò)無瓶頸的情況下，項目實測DTS全量階段的RPS最高可達30萬/s，增量階段最高可達1.5萬/s。

穩(wěn)定性強：DTS可提供高SLA保證，任務(wù)穩(wěn)定性極強。

運維成本低：用戶之前使用Canal組件時，平均每月大概需要半個人力投入到研發(fā)和運維中，改用DTS后，任務(wù)配置完成后基本無需運維人員投入，大大減少運維成本。

DTS提供的同步到Kafka數(shù)據(jù)集成方案具有通用性，目前已成功應(yīng)用在出行、零售、游戲、互聯(lián)網(wǎng)、金融等多個行業(yè)，并收獲了用戶的良好口碑。

總結(jié)和展望

DTS目前已上線了MySQL系列數(shù)據(jù)庫同步到kafka的鏈路，為用戶在大數(shù)據(jù)集成中提供了便捷的技術(shù)通道，后續(xù)為了滿足用戶更多的需求和更高的使用體驗，DTS將聚焦「數(shù)據(jù)庫生態(tài)」和「產(chǎn)品體驗」上持續(xù)發(fā)力。

數(shù)據(jù)庫生態(tài)方面：持續(xù)拓寬數(shù)據(jù)庫生態(tài)，支持其他類型的數(shù)據(jù)庫同步到kafka，如MongoDB，Oracle，PostgreSQL等同步到kafka。

產(chǎn)品體驗方面：支持更多高階特性，如全量階段支持數(shù)據(jù)可重入，投遞到多topic的策略優(yōu)化等等。