《如何基于日志，同步實現數據的一致性和實時抽取?》要點：
本文介紹了如何基于日志，同步實現數據的一致性和實時抽取?，希望對您有用。如果有疑問，可以聯系我們。

作者：王東

宜信技術研發中心架構師

• 目前就職于宜信技術研發中心,任架構師,負責流式計算和大數據業務產品解決方案.
• 曾任職于Naver china(韓國最大搜索引擎公司)中國研發中心資深工程師,多年從事CUBRID分布式數據庫集群開發和CUBRID數據庫引擎開發

  http://www.cubrid.org/blog/news/cubrid-cluster-introduction/

主題簡介：

1. DWS的背景介紹
2. dbus+wormhole總體架構和技術實現方案
3. DWS的實際運用案例

前言

大家好,我是王東,來自宜信技術研發中心,這是我來社群的第一次分享,如果有什么不足,請大家多多指正、包涵.

本次分享的主題是《基于日志的DWS平臺實現和應用》,主要是分享一下目前我們在宜信做的一些事情.這個主題里面包含到2個團隊很多兄弟姐妹的努力的結果(我們團隊和山巍團隊的成果).這次就由我代為執筆,盡我努力給大家介紹一下.

其實整個實現從原理上來說是比較簡單的,當然也涉及到不少技術.我會嘗試用盡量簡單的方式來表達,讓大家了解這個事情的原理和意義.在過程中,大家有問題可以隨時提出,我會盡力去解答.

DWS是一個簡稱,是由3個子項目組成,我稍后做解釋.

一、背景

事情是從公司前段時間的需求說起,大家知道宜信是一個互聯網金融企業,我們的很多數據與標準互聯網企業不同,大致來說就是：

玩數據的人都知道數據是非常有價值的,然后這些數據是保存在各個系統的數據庫中,如何讓需要數據的使用方得到一致性、實時的數據呢?

過去的通用做法有幾種是：

1. DBA開放各個系統的備庫,在業務低峰期(比如夜間),使用方各自抽取所需數據.由于抽取時間不同,各個數據使用方數據不一致,數據發生沖突,而且重復抽取,相信不少DBA很頭疼這個事情.
2. 公司統一的大數據平臺,通過Sqoop 在業務低峰期到各個系統統一抽取數據, 并保存到Hive表中, 然后為其他數據使用方提供數據服務.這種做法解決了一致性問題,但時效性差,基本是T+1的時效.
3. 基于trigger的方式獲取增量變更,主要問題是業務方侵入性大,而且trigger也帶來性能損失.

這些方案都不算完美.我們在了解和考慮了不同實現方式后,最后借鑒了?linkedin的思想,認為要想同時解決數據一致性和實時性,比較合理的方法應該是來自于log.

(此圖來自：https://www.confluent.io/blog/using-logs-to-build-a-solid-data-infrastructure-or-why-dual-writes-are-a-bad-idea/)

把增量的Log作為一切系統的基礎.后續的數據使用方,通過訂閱kafka來消費log.

比如：

• 大數據的使用方可以將數據保存到Hive表或者Parquet文件給Hive或Spark查詢；
• 提供搜索服務的使用方可以保存到Elasticsearch或HBase 中；
• 提供緩存服務的使用方可以將日志緩存到Redis或alluxio中；
• 數據同步的使用方可以將數據保存到自己的數據庫中；
• 由于kafka的日志是可以重復消費的,并且緩存一段時間,各個使用方可以通過消費kafka的日志來達到既能保持與數據庫的一致性,也能保證實時性；

為什么使用log和kafka作為基礎,而不使用Sqoop進行抽取呢? 因為：

為什么不使用dual write(雙寫)呢?,請參考https://www.confluent.io/blog/using-logs-to-build-a-solid-data-infrastructure-or-why-dual-writes-are-a-bad-idea/

我這里就不多做解釋了.

二、總體架構

于是我們提出了構建一個基于log的公司級的平臺的想法.

下面解釋一下DWS平臺, DWS平臺是有3個子項目組成：

1. Dbus(數據總線)：負責實時將數據從源端實時抽出,并轉換為約定的自帶schema的json格式數據(UMS?數據),放入kafka中；
2. Wormhole(數據交換平臺)：負責從kafka讀出數據 將數據寫入到目標中；
3. Swifts(實時計算平臺)：負責從kafka中讀出數據,實時計算,并將數據寫回kafka中.

圖中：

• Log extractor和dbus共同完成數據抽取和數據轉換,抽取包括全量和增量抽取.
• Wormhole可以將所有日志數據保存到HDFS中； 還可以將數據落地到所有支持jdbc的數據庫,落地到HBash,Elasticsearch,Cassandra等；
• Swifts支持以配置和SQL的方式實現對進行流式計算,包括支持流式join,look up,filter,window aggregation等功能；
• Dbus web是dbus的配置管理端,rider除了配置管理以外,還包括對Wormhole和Swifts運行時管理,數據質量校驗等.

由于時間關系,我今天主要介紹DWS中的Dbus和Wormhole,在需要的時候附帶介紹一下Swifts.

三、dbus解決方案

日志解析

如前面所說,Dbus主要解決的是將日志從源端實時的抽出. 這里我們以MySQL為例子,簡單說明如何實現.

我們知道,雖然MySQL InnoDB有自己的log,MySQL主備同步是通過binlog來實現的.如下圖：

圖片來自：https://github.com/alibaba/canal

而binlog有三種模式：

1. Row 模式：日志中會記錄成每一行數據被修改的形式,然后在slave端再對相同的數據進行修改.
2. Statement 模式: 每一條會修改數據的sql都會記錄到 master的bin-log中.slave在復制的時候SQL進程會解析成和原來master端執行過的相同的SQL來再次執行.
3. Mixed模式： MySQL會根據執行的每一條具體的sql語句來區分對待記錄的日志形式,也就是在Statement和Row之間選擇一種.

他們各自的優缺點如下：

此處來自：http://www.jquerycn.cn/a_13625

由于statement 模式的缺點,在與我們的DBA溝通過程中了解到,實際生產過程中都使用row 模式進行復制.這使得讀取全量日志成為可能.

通常我們的MySQL布局是采用 2個master主庫(vip)+ 1個slave從庫 + 1個backup容災庫 的解決方案,由于容災庫通常是用于異地容災,實時性不高也不便于部署.

為了最小化對源端產生影響,顯然我們讀取binlog日志應該從slave從庫讀取.

讀取binlog的方案比較多,github上不少,參考https://github.com/search?utf8=%E2%9C%93&q=binlog.最終我們選用了阿里的canal做位日志抽取方.

Canal最早被用于阿里中美機房同步, canal原理相對比較簡單：

1. Canal模擬MySQL Slave的交互協議,偽裝自己為MySQL Slave,向MySQL Slave發送dump協議
2. MySQL master收到dump請求,開始推送binary log給Slave(也就是canal)
3. Canal解析binary log對象(原始為byte流)

圖片來自：https://github.com/alibaba/canal

解決方案

Dbus 的MySQL版主要解決方案如下：

對于增量的log,通過訂閱Canal Server的方式,我們得到了MySQL的增量日志：

• 按照Canal的輸出,日志是protobuf格式,開發增量Storm程序,將數據實時轉換為我們定義的UMS格式(json格式,稍后我會介紹),并保存到kafka中；
• 增量Storm程序還負責捕獲schema變化,以控制版本號；
• 增量Storm的配置信息保存在Zookeeper中,以滿足高可用需求.
• Kafka既作為輸出結果也作為處理過程中的緩沖器和消息解構區.

在考慮使用Storm作為解決方案的時候,我們主要是認為Storm有以下優點：

• 技術相對成熟,比較穩定,與kafka搭配也算標準組合；
• 實時性比較高,能夠滿足實時性需求；
• 滿足高可用需求；
• 通過配置Storm并發度,可以活動性能擴展的能力；

全量抽取

對于流水表,有增量部分就夠了,但是許多表需要知道最初(已存在)的信息.這時候我們需要initial load(第一次加載).

對于initial load(第一次加載),同樣開發了全量抽取Storm程序通過jdbc連接的方式,從源端數據庫的備庫進行拉取.initial load是拉全部數據,所以我們推薦在業務低峰期進行.好在只做一次,不需要每天都做.

全量抽取,我們借鑒了Sqoop的思想.將全量抽取Storm分為了2 個部分：

1. 數據分片
2. 實際抽取

數據分片需要考慮分片列,按照配置和自動選擇列將數據按照范圍來分片,并將分片信息保存到kafka中.

下面是具體的分片策略：

全量抽取的Storm程序是讀取kafka的分片信息,采用多個并發度并行連接數據庫備庫進行拉取.因為抽取的時間可能很長.抽取過程中將實時狀態寫到Zookeeper中,便于心跳程序監控.

統一消息格式

無論是增量還是全量,最終輸出到kafka中的消息都是我們約定的一個統一消息格式,稱為UMS(unified message schema)格式.

如下圖所示：

消息中schema部分,定義了namespace 是由 類型+數據源名+schema名+表名+版本號+分庫號+分表號?能夠描述整個公司的所有表,通過一個namespace就能唯一定位.

• _ums_op_ 表明數據的類型是I(insert),U(update),D(刪除)；
• _ums_ts_ 發生增刪改的事件的時間戳,顯然新的數據發生的時間戳更新；
• _ums_id_ 消息的唯一id,保證消息是唯一的,但這里我們保證了消息的先后順序(稍后解釋)；

payload是指具體的數據,一個json包里面可以包含1條至多條數據,提高數據的有效載荷.

UMS中支持的數據類型,參考了Hive類型并進行簡化,基本上包含了所有數據類型.

全量和增量的一致性

在整個數據傳輸中,為了盡量的保證日志消息的順序性,kafka我們使用的是1個partition的方式.在一般情況下,基本上是順序的和唯一的.

但是我們知道寫kafka會失敗,有可能重寫,Storm也用重做機制,因此,我們并不嚴格保證exactly once和完全的順序性,但保證的是at least once.

因此_ums_id_變得尤為重要.

對于全量抽取,_ums_id_是唯一的,從zk中每個并發度分別取不同的id片區,保證了唯一性和性能,填寫負數,不會與增量數據沖突,也保證他們是早于增量消息的.

對于增量抽取,我們使用的是MySQL的日志文件號 + 日志偏移量作為唯一id.Id作為64位的long整數,高7位用于日志文件號,低12位作為日志偏移量.

例如：000103000012345678. 103 是日志文件號,12345678 是日志偏移量.

這樣,從日志層面保證了物理唯一性(即便重做也這個id號也不變),同時也保證了順序性(還能定位日志).通過比較_ums_id_ 消費日志就能通過比較_ums_id_知道哪條消息更新.

其實_ums_ts_與_ums_id_意圖是類似的,只不過有時候_ums_ts_可能會重復,即在1毫秒中發生了多個操作,這樣就得靠比較_ums_id_了.

心跳監控和預警

整個系統涉及到數據庫的主備同步,Canal Server,多個并發度Storm進程等各個環節.

因此對流程的監控和預警就尤為重要.

通過心跳模塊,例如每分鐘(可配置)對每個被抽取的表插入一條心態數據并保存發送時間,這個心跳表也被抽取,跟隨著整個流程下來,與被同步表在實際上走相同的邏輯(因為多個并發的的Storm可能有不同的分支),當收到心跳包的時候,即便沒有任何增刪改的數據,也能證明整條鏈路是通的.

Storm程序和心跳程序將數據發送公共的統計topic,再由統計程序保存到influxdb中,使用grafana進行展示,就可以看到如下效果：

圖中是某業務系統的實時監控信息.上面是實時流量情況,下面是實時延時情況.可以看到,實時性還是很不錯的,基本上1~2秒數據就已經到末端kafka中.

Granfana提供的是一種實時監控能力.

如果出現延時,則是通過dbus的心跳模塊發送郵件報警或短信報警.

實時脫敏

考慮到數據安全性,對于有脫敏需求的場景,Dbus的全量storm和增量storm程序也完成了實時脫敏的功能.脫敏方式有3種：

總結一下：簡單的說,Dbus就是將各種源的數據,實時的導出,并以UMS的方式提供訂閱, 支持實時脫敏,實際監控和報警.

四、Wormhole解決方案

說完Dbus,該說一下Wormhole,為什么兩個項目不是一個,而要通過kafka來對接呢?

其中很大一個原因就是解耦,kafka具有天然的解耦能力,程序直接可以通過kafka做異步的消息傳遞.Dbus和Wornhole內部也使用了kafka做消息傳遞和解耦.

另外一個原因就是,UMS是自描述的,通過訂閱kafka,任何有能力的使用方來直接消費UMS來使用.

雖然UMS的結果可以直接訂閱,但還需要開發的工作.Wormhole解決的是：提供一鍵式的配置,將kafka中的數據落地到各種系統中,讓沒有開發能力的數據使用方通過wormhole來實現使用數據.

如圖所示,Wormhole 可以將kafka中的UMS 落地到各種系統,目前用的最多的HDFS,JDBC的數據庫和HBase.

在技術棧上, wormhole選擇使用spark streaming來進行.

在Wormhole中,一條flow是指從一個namaspace從源端到目標端.一個spark streaming服務于多條flow.

選用Spark的理由是很充分的：

• Spark天然的支持各種異構存儲系統；
• 雖然Spark Stream比Storm延時稍差,但Spark有著更好的吞吐量和更好的計算性能；
• Spark在支持并行計算方面有更強的靈活性；
• Spark提供了一個技術棧內解決Sparking Job,Spark Streaming,Spark SQL的統一功能,便于后期開發；

這里補充說一下Swifts的作用：

• Swifts的本質是讀取kafka中的UMS數據,進行實時計算,將結果寫入到kafka的另外一個topic.
• 實時計算可以是很多種方式：比如過濾filter,projection(投影),lookup, 流式join window aggregation,可以完成各種具有業務價值的流式實時計算.

Wormhole和Swifts對比如下：

落HDFS

通過Wormhole Wpark Streaming程序消費kafka的UMS,首先UMS log可以被保存到HDFS上.

kafka一般只保存若干天的信息,不會保存全部信息,而HDFS中可以保存所有的歷史增刪改的信息.這就使得很多事情變為可能：

• 通過重放HDFS中的日志,我們能夠還原任意時間的歷史快照.
• 可以做拉鏈表,還原每一條記錄的歷史信息,便于分析；
• 當程序出現錯誤是,可以通過回灌(backfill),重新消費消息,重新形成新的快照.

可以說HDFS中的日志是很多的事情基礎.

介于Spark原生對parquet支持的很好,Spark SQL能夠對Parquet提供很好的查詢.UMS落地到HDFS上是保存到Parquet文件中的.Parquet的內容是所有log的增刪改信息以及_ums_id_,_ums_ts_都存下來.

Wormhole spark streaming根據namespace 將數據分布存儲到不同的目錄中,即不同的表和版本放在不同目錄中.

由于每次寫的Parquet都是小文件,大家知道HDFS對于小文件性能并不好,因此另外還有一個job,每天定時將這些的Parquet文件進行合并成大文件.

每個Parquet文件目錄都帶有文件數據的起始時間和結束時間.這樣在回灌數據時,可以根據選取的時間范圍來決定需要讀取哪些Parquet文件,不必讀取全部數據.

插入或更新數據的冪等性

常常我們遇到的需求是,將數據經過加工落地到數據庫或HBase中.那么這里涉及到的一個問題就是,什么樣的數據可以被更新到數據?

這里最重要的一個原則就是數據的冪等性.

無論是遇到增刪改任何的數據,我們面臨的問題都是：

1. 該更新哪一行；
2. 更新的策略是什么.

對于第一個問題,其實就需要定位數據要找一個唯一的鍵,常見的有：

1. 使用業務庫的主鍵；
2. 由業務方指定幾個列做聯合唯一索引；

對于第二個問題,就涉及到_ums_id_了,因為我們已經保證了_ums_id_大的值更新,因此在找到對應數據行后,根據這個原則來進行替換更新.

之所以要軟刪除和加入_is_active_列,是為了這樣一種情況：

如果已經插入的_ums_id_比較大,是刪除的數據(表明這個數據已經刪除了), 如果不是軟刪除,此時插入一個_ums_id_小的數據(舊數據),就會真的插入進去.

這就導致舊數據被插入了.不冪等了.所以被刪除的數據依然保留(軟刪除)是有價值的,它能被用于保證數據的冪等性.

HBase的保存

插入數據到Hbase中,相當要簡單一些.不同的是HBase可以保留多個版本的數據(當然也可以只保留一個版本)默認是保留3個版本；

因此插入數據到HBase,需要解決的問題是：

1. 選擇合適的rowkey：Rowkey的設計是可以選的,用戶可以選擇源表的主鍵,也可以選擇若干列做聯合主鍵.
2. 選擇合適的version：使用_ums_id_+ 較大的偏移量(比如100億) 作為row的version.

Version的選擇很有意思,利用_ums_id_的唯一性和自增性,與version自身的比較關系一致：即version較大等價于_ums_id_較大,對應的版本較新.

從提高性能的角度,我們可以將整個Spark Streaming的Dataset集合直接插入到HBase,不需要比較.讓HBase基于version自動替我們判斷哪些數據可以保留,哪些數據不需要保留.

Jdbc的插入數據：

插入數據到數據庫中,保證冪等的原理雖然簡單,要想提高性能在實現上就變得復雜很多,總不能一條一條的比較然后在插入或更新.

我們知道Spark的RDD/dataset都是以集合的方式來操作以提高性能,同樣的我們需要以集合操作的方式實現冪等性.

具體思路是：

1. 首先根據集合中的主鍵到目標數據庫中查詢,得到一個已有數據集合；
2. 與dataset中的集合比較,分出兩類：

A：不存在的數據,即這部分數據insert就可以；

B：存在的數據,比較_ums_id_, 最終只將哪些_ums_id_更新較大row到目標數據庫,小的直接拋棄.

使用Spark的同學都知道,RDD/dataset都是可以partition的,可以使用多個worker并進行操作以提高效率.

在考慮并發情況下,插入和更新都可能出現失敗,那么還有考慮失敗后的策略.

比如：因為別的worker已經插入,那么因為唯一性約束插入失敗,那么需要改為更新,還要比較_ums_id_看是否能夠更新.

對于無法插入其他情況(比如目標系統有問題),Wormhole還有重試機制.說起來細節特別多.這里就不多介紹了.

有些還在開發中.

插入到其他存儲中的就不多介紹了,總的原則是：根據各自存儲自身特性,設計基于集合的,并發的插入數據實現.這些都是Wormhole為了性能而做的努力,使用Ｗormhole的用戶不必關心 .

五、運用案例

實時營銷

說了那么多,DWS有什么實際運用呢?下面我來介紹某系統使用DWS實現了的實時營銷.

如上圖所示：

系統A的數據都保存到自己的數據庫中,我們知道,宜信提供很多金融服務,其中包括借款,而借款過程中很重要的就是信用審核.

借款人需要提供證明具有信用價值的信息,比如央行征信報告,是具有最強信用數據的數據. 而銀行流水,網購流水也是具有較強的信用屬性的數據.

借款人通過Web或手機APP在系統A中填寫信用信息時,可能會某些原因無法繼續,雖然可能這個借款人是一個優質潛在客戶,但以前由于無法或很久才能知道這個信息,所以實際上這樣的客戶是流失了.

應用了DWS以后,借款人已經填寫的信息已經記錄到數據庫中,并通過DWS實時的進行抽取、計算和落地到目標庫中.根據對客戶的打分,評價出優質客戶.然后立刻將這個客戶的信息輸出到客服系統中.

客服人員在很短的時間(幾分鐘以內)就通過打電話的方式聯系上這個借款人(潛客),進行客戶關懷,將這個潛客轉換為真正的客戶.我們知道借款是有時效性的,如果時間太久就沒有價值了.

如果沒有實時抽取/計算/落庫的能力,那么這一切都無法實現.

實時報表系統

另外一個實時報表的應用如下：

我們數據使用方的數據來自多個系統,以前是通過T+1的方式獲得報表信息,然后指導第二天的運營,這樣時效性很差.

通過DWS,將數據從多個系統中實時抽取,計算和落地,并提供報表展示,使得運營可以及時作出部署和調整,快速應對.

六、總結

說了那么多,大致總結一下：

• DWS技術上基于主流實時流式大數據技術框架,高可用大吞吐強水平擴容,低延遲高容錯最終一致.
• DWS能力上支持異構多源多目標系統,支持多數據格式(結構化半結構化非結構化數據)和實時技術能力.
• DWS將三個子項目合并作為一個平臺推出,使得我們具備了實時的能力, 驅動各種實時場景應用.

適合場景包括：實時同步／實時計算／實時監控／實時報表／實時分析／實時洞察／實時管理／實時運營／實時決策

感謝大家的聆聽,此次分享到此為止.

Q&A

Q1：Oracle log reader有開源方案嗎?

Ａ1：對于Oracle業界也有許多商業解決方案,例如：Oracle GoldenGate(原來的goldengate), Oracle Xstream, IBM InfoSphere Change Data Capture(原來的DataMirror),Dell SharePlex (原來的Quest),國內的DSG superSync等,開源的方案好用的很少.

Q2：這個項目投入了多少人力物力?感覺有點復雜.

Q2：DWS是三個子項目組成,平均每個項目5~7人.是有點復雜,其實也是試圖使用大數據技術來解決我們公司目前遇到的困難.

因為是搞大數據相關技術,所有團隊里面的兄弟姐妹都還是比較happy的：)

其實這里面,Dbus和Wormhole相對固定模式化,容易輕松復用.Swifts實時計算是與每個業務相關比較大的,自定義比較強,相對比較麻煩一些.

Q3：宜信的這個DWS系統會開源么?

A3：我們也考慮過向社區貢獻,就像宜信的其他開源項目一樣,目前項目剛剛成形,還有待進一步磨煉,我相信未來的某個時候,我們會給它開源出來.

Q4：架構師怎么理解,是不是系統工程師?

A4：不是系統工程師,在我們宜信有多位架構師,應該算是以技術驅動業務的技術管理人員.包含產品設計,技術管理等.

Q5：復制方案是否是OGG?

A5：OGG與上面提到的其他商業解決方案都是可選方案.

文章出處：DBAplus社群(dbaplus)

轉載請注明本頁網址：
http://www.fzlkiss.com/jiaocheng/4376.html