DB數(shù)據(jù)同步到數(shù)據(jù)倉(cāng)庫(kù)的示例分析

這篇文章給大家分享的是有關(guān)DB數(shù)據(jù)同步到數(shù)據(jù)倉(cāng)庫(kù)的示例分析的內(nèi)容。小編覺(jué)得挺實(shí)用的，因此分享給大家做個(gè)參考，一起跟隨小編過(guò)來(lái)看看吧。

背景

在數(shù)據(jù)倉(cāng)庫(kù)建模中，未經(jīng)任何加工處理的原始業(yè)務(wù)層數(shù)據(jù)，我們稱之為ODS(Operational Data Store)數(shù)據(jù)。在互聯(lián)網(wǎng)企業(yè)中，常見(jiàn)的ODS數(shù)據(jù)有業(yè)務(wù)日志數(shù)據(jù)（Log）和業(yè)務(wù)DB數(shù)據(jù)（DB）兩類。對(duì)于業(yè)務(wù)DB數(shù)據(jù)來(lái)說(shuō)，從MySQL等關(guān)系型數(shù)據(jù)庫(kù)的業(yè)務(wù)數(shù)據(jù)進(jìn)行采集，然后導(dǎo)入到Hive中，是進(jìn)行數(shù)據(jù)倉(cāng)庫(kù)生產(chǎn)的重要環(huán)節(jié)。

如何準(zhǔn)確、高效地把MySQL數(shù)據(jù)同步到Hive中？一般常用的解決方案是批量取數(shù)并Load：直連MySQL去Select表中的數(shù)據(jù)，然后存到本地文件作為中間存儲(chǔ)，最后把文件Load到Hive表中。這種方案的優(yōu)點(diǎn)是實(shí)現(xiàn)簡(jiǎn)單，但是隨著業(yè)務(wù)的發(fā)展，缺點(diǎn)也逐漸暴露出來(lái)：

• 性能瓶頸：隨著業(yè)務(wù)規(guī)模的增長(zhǎng)，Select From MySQL -> Save to Localfile -> Load to Hive這種數(shù)據(jù)流花費(fèi)的時(shí)間越來(lái)越長(zhǎng)，無(wú)法滿足下游數(shù)倉(cāng)生產(chǎn)的時(shí)間要求。

• 直接從MySQL中Select大量數(shù)據(jù)，對(duì)MySQL的影響非常大，容易造成慢查詢，影響業(yè)務(wù)線上的正常服務(wù)。

• 由于Hive本身的語(yǔ)法不支持更新、刪除等SQL原語(yǔ)，對(duì)于MySQL中發(fā)生Update/Delete的數(shù)據(jù)無(wú)法很好地進(jìn)行支持。

為了徹底解決這些問(wèn)題，我們逐步轉(zhuǎn)向CDC (Change Data Capture) + Merge的技術(shù)方案，即實(shí)時(shí)Binlog采集 + 離線處理Binlog還原業(yè)務(wù)數(shù)據(jù)這樣一套解決方案。Binlog是MySQL的二進(jìn)制日志，記錄了MySQL中發(fā)生的所有數(shù)據(jù)變更，MySQL集群自身的主從同步就是基于Binlog做的。

本文主要從Binlog實(shí)時(shí)采集和離線處理Binlog還原業(yè)務(wù)數(shù)據(jù)兩個(gè)方面，來(lái)介紹如何實(shí)現(xiàn)DB數(shù)據(jù)準(zhǔn)確、高效地進(jìn)入數(shù)倉(cāng)。

整體架構(gòu)

整體的架構(gòu)如上圖所示。在Binlog實(shí)時(shí)采集方面，我們采用了阿里巴巴的開(kāi)源項(xiàng)目Canal，負(fù)責(zé)從MySQL實(shí)時(shí)拉取Binlog并完成適當(dāng)解析。Binlog采集后會(huì)暫存到Kafka上供下游消費(fèi)。整體實(shí)時(shí)采集部分如圖中紅色箭頭所示。

離線處理Binlog的部分，如圖中黑色箭頭所示，通過(guò)下面的步驟在Hive上還原一張MySQL表：

• 采用Linkedin的開(kāi)源項(xiàng)目Camus，負(fù)責(zé)每小時(shí)把Kafka上的Binlog數(shù)據(jù)拉取到Hive上。

• 對(duì)每張ODS表，首先需要一次性制作快照（Snapshot），把MySQL里的存量數(shù)據(jù)讀取到Hive上，這一過(guò)程底層采用直連MySQL去Select數(shù)據(jù)的方式。

• 對(duì)每張ODS表，每天基于存量數(shù)據(jù)和當(dāng)天增量產(chǎn)生的Binlog做Merge，從而還原出業(yè)務(wù)數(shù)據(jù)。

我們回過(guò)頭來(lái)看看，背景中介紹的批量取數(shù)并Load方案遇到的各種問(wèn)題，為什么用這種方案能解決上面的問(wèn)題呢？

• 首先，Binlog是流式產(chǎn)生的，通過(guò)對(duì)Binlog的實(shí)時(shí)采集，把部分?jǐn)?shù)據(jù)處理需求由每天一次的批處理分?jǐn)偟綄?shí)時(shí)流上。無(wú)論從性能上還是對(duì)MySQL的訪問(wèn)壓力上，都會(huì)有明顯地改善。

• 第二，Binlog本身記錄了數(shù)據(jù)變更的類型（Insert/Update/Delete），通過(guò)一些語(yǔ)義方面的處理，完全能夠做到精準(zhǔn)的數(shù)據(jù)還原。

Binlog實(shí)時(shí)采集

對(duì)Binlog的實(shí)時(shí)采集包含兩個(gè)主要模塊：一是CanalManager，主要負(fù)責(zé)采集任務(wù)的分配、監(jiān)控報(bào)警、元數(shù)據(jù)管理以及和外部依賴系統(tǒng)的對(duì)接；二是真正執(zhí)行采集任務(wù)的Canal和CanalClient。

當(dāng)用戶提交某個(gè)DB的Binlog采集請(qǐng)求時(shí)，CanalManager首先會(huì)調(diào)用DBA平臺(tái)的相關(guān)接口，獲取這一DB所在MySQL實(shí)例的相關(guān)信息，目的是從中選出最適合Binlog采集的機(jī)器。然后把采集實(shí)例（Canal Instance）分發(fā)到合適的Canal服務(wù)器上，即CanalServer上。在選擇具體的CanalServer時(shí)，CanalManager會(huì)考慮負(fù)載均衡、跨機(jī)房傳輸?shù)纫蛩?，?yōu)先選擇負(fù)載較低且同地域傳輸?shù)臋C(jī)器。

CanalServer收到采集請(qǐng)求后，會(huì)在ZooKeeper上對(duì)收集信息進(jìn)行注冊(cè)。注冊(cè)的內(nèi)容包括：

• 以Instance名稱命名的永久節(jié)點(diǎn)。

• 在該永久節(jié)點(diǎn)下注冊(cè)以自身ip:port命名的臨時(shí)節(jié)點(diǎn)。

這樣做的目的有兩個(gè)：

• 高可用：CanalManager對(duì)Instance進(jìn)行分發(fā)時(shí)，會(huì)選擇兩臺(tái)CanalServer，一臺(tái)是Running節(jié)點(diǎn)，另一臺(tái)作為Standby節(jié)點(diǎn)。Standby節(jié)點(diǎn)會(huì)對(duì)該Instance進(jìn)行監(jiān)聽(tīng)，當(dāng)Running節(jié)點(diǎn)出現(xiàn)故障后，臨時(shí)節(jié)點(diǎn)消失，然后Standby節(jié)點(diǎn)進(jìn)行搶占。這樣就達(dá)到了容災(zāi)的目的。

• 與CanalClient交互：CanalClient檢測(cè)到自己負(fù)責(zé)的Instance所在的Running CanalServer后，便會(huì)進(jìn)行連接，從而接收到CanalServer發(fā)來(lái)的Binlog數(shù)據(jù)。

對(duì)Binlog的訂閱以MySQL的DB為粒度，一個(gè)DB的Binlog對(duì)應(yīng)了一個(gè)Kafka Topic。底層實(shí)現(xiàn)時(shí)，一個(gè)MySQL實(shí)例下所有訂閱的DB，都由同一個(gè)Canal Instance進(jìn)行處理。這是因?yàn)锽inlog的產(chǎn)生是以MySQL實(shí)例為粒度的。CanalServer會(huì)拋棄掉未訂閱的Binlog數(shù)據(jù)，然后CanalClient將接收到的Binlog按DB粒度分發(fā)到Kafka上。

離線還原MySQL數(shù)據(jù)

完成Binlog采集后，下一步就是利用Binlog來(lái)還原業(yè)務(wù)數(shù)據(jù)。首先要解決的第一個(gè)問(wèn)題是把Binlog從Kafka同步到Hive上。

Kafka2Hive

整個(gè)Kafka2Hive任務(wù)的管理，在美團(tuán)數(shù)據(jù)平臺(tái)的ETL框架下進(jìn)行，包括任務(wù)原語(yǔ)的表達(dá)和調(diào)度機(jī)制等，都同其他ETL類似。而底層采用LinkedIn的開(kāi)源項(xiàng)目Camus，并進(jìn)行了有針對(duì)性的二次開(kāi)發(fā)，來(lái)完成真正的Kafka2Hive數(shù)據(jù)傳輸工作。

對(duì)Camus的二次開(kāi)發(fā)

Kafka上存儲(chǔ)的Binlog未帶Schema，而Hive表必須有Schema，并且其分區(qū)、字段等的設(shè)計(jì)，都要便于下游的高效消費(fèi)。對(duì)Camus做的第一個(gè)改造，便是將Kafka上的Binlog解析成符合目標(biāo)Schema的格式。

對(duì)Camus做的第二個(gè)改造，由美團(tuán)的ETL框架所決定。在我們的任務(wù)調(diào)度系統(tǒng)中，目前只對(duì)同調(diào)度隊(duì)列的任務(wù)做上下游依賴關(guān)系的解析，跨調(diào)度隊(duì)列是不能建立依賴關(guān)系的。而在MySQL2Hive的整個(gè)流程中，Kafka2Hive的任務(wù)需要每小時(shí)執(zhí)行一次（小時(shí)隊(duì)列），Merge任務(wù)每天執(zhí)行一次（天隊(duì)列）。而Merge任務(wù)的啟動(dòng)必須要嚴(yán)格依賴小時(shí)Kafka2Hive任務(wù)的完成。

為了解決這一問(wèn)題，我們引入了Checkdone任務(wù)。Checkdone任務(wù)是天任務(wù)，主要負(fù)責(zé)檢測(cè)前一天的Kafka2Hive是否成功完成。如果成功完成了，則Checkdone任務(wù)執(zhí)行成功，這樣下游的Merge任務(wù)就可以正確啟動(dòng)了。

Checkdone的檢測(cè)邏輯

Checkdone是怎樣檢測(cè)的呢？每個(gè)Kafka2Hive任務(wù)成功完成數(shù)據(jù)傳輸后，由Camus負(fù)責(zé)在相應(yīng)的HDFS目錄下記錄該任務(wù)的啟動(dòng)時(shí)間。Checkdone會(huì)掃描前一天的所有時(shí)間戳，如果最大的時(shí)間戳已經(jīng)超過(guò)了0點(diǎn)，就說(shuō)明前一天的Kafka2Hive任務(wù)都成功完成了，這樣Checkdone就完成了檢測(cè)。

此外，由于Camus本身只是完成了讀Kafka然后寫(xiě)HDFS文件的過(guò)程，還必須完成對(duì)Hive分區(qū)的加載才能使下游查詢到。因此，整個(gè)Kafka2Hive任務(wù)的最后一步是加載Hive分區(qū)。這樣，整個(gè)任務(wù)才算成功執(zhí)行。

每個(gè)Kafka2Hive任務(wù)負(fù)責(zé)讀取一個(gè)特定的Topic，把Binlog數(shù)據(jù)寫(xiě)入original_binlog庫(kù)下的一張表中，即前面圖中的original_binlog.db，其中存儲(chǔ)的是對(duì)應(yīng)到一個(gè)MySQL DB的全部Binlog。

上圖說(shuō)明了一個(gè)Kafka2Hive完成后，文件在HDFS上的目錄結(jié)構(gòu)。假如一個(gè)MySQL DB叫做user，對(duì)應(yīng)的Binlog存儲(chǔ)在original_binlog.user表中。ready目錄中，按天存儲(chǔ)了當(dāng)天所有成功執(zhí)行的Kafka2Hive任務(wù)的啟動(dòng)時(shí)間，供Checkdone使用。每張表的Binlog，被組織到一個(gè)分區(qū)中，例如userinfo表的Binlog，存儲(chǔ)在table_name=userinfo這一分區(qū)中。每個(gè)table_name一級(jí)分區(qū)下，按dt組織二級(jí)分區(qū)。圖中的xxx.lzo和xxx.lzo.index文件，存儲(chǔ)的是經(jīng)過(guò)lzo壓縮的Binlog數(shù)據(jù)。

Merge

Binlog成功入倉(cāng)后，下一步要做的就是基于Binlog對(duì)MySQL數(shù)據(jù)進(jìn)行還原。Merge流程做了兩件事，首先把當(dāng)天生成的Binlog數(shù)據(jù)存放到Delta表中，然后和已有的存量數(shù)據(jù)做一個(gè)基于主鍵的Merge。Delta表中的數(shù)據(jù)是當(dāng)天的最新數(shù)據(jù)，當(dāng)一條數(shù)據(jù)在一天內(nèi)發(fā)生多次變更時(shí)，Delta表中只存儲(chǔ)最后一次變更后的數(shù)據(jù)。

把Delta數(shù)據(jù)和存量數(shù)據(jù)進(jìn)行Merge的過(guò)程中，需要有唯一鍵來(lái)判定是否是同一條數(shù)據(jù)。如果同一條數(shù)據(jù)既出現(xiàn)在存量表中，又出現(xiàn)在Delta表中，說(shuō)明這一條數(shù)據(jù)發(fā)生了更新，則選取Delta表的數(shù)據(jù)作為最終結(jié)果；否則說(shuō)明沒(méi)有發(fā)生任何變動(dòng)，保留原來(lái)存量表中的數(shù)據(jù)作為最終結(jié)果。Merge的結(jié)果數(shù)據(jù)會(huì)Insert Overwrite到原表中，即圖中的origindb.table。

Merge流程舉例

下面用一個(gè)例子來(lái)具體說(shuō)明Merge的流程。

數(shù)據(jù)表共id、value兩列，其中id是主鍵。在提取Delta數(shù)據(jù)時(shí)，對(duì)同一條數(shù)據(jù)的多次更新，只選擇最后更新的一條。所以對(duì)id=1的數(shù)據(jù)，Delta表中記錄最后一條更新后的值value=120。Delta數(shù)據(jù)和存量數(shù)據(jù)做Merge后，最終結(jié)果中，新插入一條數(shù)據(jù)（id=4），兩條數(shù)據(jù)發(fā)生了更新（id=1和id=2），一條數(shù)據(jù)未變（id=3）。

默認(rèn)情況下，我們采用MySQL表的主鍵作為這一判重的唯一鍵，業(yè)務(wù)也可以根據(jù)實(shí)際情況配置不同于MySQL的唯一鍵。

上面介紹了基于Binlog的數(shù)據(jù)采集和ODS數(shù)據(jù)還原的整體架構(gòu)。下面主要從兩個(gè)方面介紹我們解決的實(shí)際業(yè)務(wù)問(wèn)題。

實(shí)踐一：分庫(kù)分表的支持

隨著業(yè)務(wù)規(guī)模的擴(kuò)大，MySQL的分庫(kù)分表情況越來(lái)越多，很多業(yè)務(wù)的分表數(shù)目都在幾千個(gè)這樣的量級(jí)。而一般數(shù)據(jù)開(kāi)發(fā)同學(xué)需要把這些數(shù)據(jù)聚合到一起進(jìn)行分析。如果對(duì)每個(gè)分表都進(jìn)行手動(dòng)同步，再在Hive上進(jìn)行聚合，這個(gè)成本很難被我們接受。因此，我們需要在ODS層就完成分表的聚合。

首先，在Binlog實(shí)時(shí)采集時(shí)，我們支持把不同DB的Binlog寫(xiě)入到同一個(gè)Kafka Topic。用戶可以在申請(qǐng)Binlog采集時(shí)，同時(shí)勾選同一個(gè)業(yè)務(wù)邏輯下的多個(gè)物理DB。通過(guò)在Binlog采集層的匯集，所有分庫(kù)的Binlog會(huì)寫(xiě)入到同一張Hive表中，這樣下游在進(jìn)行Merge時(shí)，依然只需要讀取一張Hive表。

第二，Merge任務(wù)的配置支持正則匹配。通過(guò)配置符合業(yè)務(wù)分表命名規(guī)則的正則表達(dá)式，Merge任務(wù)就能了解自己需要聚合哪些MySQL表的Binlog，從而選取相應(yīng)分區(qū)的數(shù)據(jù)來(lái)執(zhí)行。

這樣通過(guò)兩個(gè)層面的工作，就完成了分庫(kù)分表在ODS層的合并。

這里面有一個(gè)技術(shù)上的優(yōu)化，在進(jìn)行Kafka2Hive時(shí)，我們按業(yè)務(wù)分表規(guī)則對(duì)表名進(jìn)行了處理，把物理表名轉(zhuǎn)換成了邏輯表名。例如userinfo123這張表名會(huì)被轉(zhuǎn)換為userinfo，其Binlog數(shù)據(jù)存儲(chǔ)在original_binlog.user表的table_name=userinfo分區(qū)中。這樣做的目的是防止過(guò)多的HDFS小文件和Hive分區(qū)造成的底層壓力。

實(shí)踐二：刪除事件的支持

Delete操作在MySQL中非常常見(jiàn)，由于Hive不支持Delete，如果想把MySQL中刪除的數(shù)據(jù)在Hive中刪掉，需要采用“迂回”的方式進(jìn)行。

對(duì)需要處理Delete事件的Merge流程，采用如下兩個(gè)步驟：

• 首先，提取出發(fā)生了Delete事件的數(shù)據(jù)，由于Binlog本身記錄了事件類型，這一步很容易做到。將存量數(shù)據(jù)（表A）與被刪掉的數(shù)據(jù)（表B）在主鍵上做左外連接(Left outer join)，如果能夠全部join到雙方的數(shù)據(jù)，說(shuō)明該條數(shù)據(jù)被刪掉了。因此，選擇結(jié)果中表B對(duì)應(yīng)的記錄為NULL的數(shù)據(jù)，即是應(yīng)當(dāng)被保留的數(shù)據(jù)。

• 然后，對(duì)上面得到的被保留下來(lái)的數(shù)據(jù)，按照前面描述的流程做常規(guī)的Merge。

感謝各位的閱讀！關(guān)于“DB數(shù)據(jù)同步到數(shù)據(jù)倉(cāng)庫(kù)的示例分析”這篇文章就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，讓大家可以學(xué)到更多知識(shí)，如果覺(jué)得文章不錯(cuò)，可以把它分享出去讓更多的人看到吧！