Spark實時統(tǒng)計訂單量 - 簡書
http://www.itdecent.cn/p/3ec093a9d584

Paste_Image.png

注：組件不了解的同學可參考其他文章，本文主要講項目的實現(xiàn)
1、某些同學會問，直接在業(yè)務系統(tǒng)加入JS埋點通過發(fā)日志不更好嗎？
答：第一、JS埋點業(yè)務系統(tǒng)涉及產(chǎn)品改造，不可能因為一個需求讓你去隨便改業(yè)務系統(tǒng)。第二、即使加入JS埋點也不可能獲得業(yè)務系統(tǒng)的全部數(shù)據(jù)。所以業(yè)務系統(tǒng)核心數(shù)據(jù)還得去業(yè)務系統(tǒng)庫獲取。

2、還有人問加入Kafka太多余
答：第一、加入Kafka為了使系統(tǒng)擴展性更強，可方便對接各種開源產(chǎn)品。第二、通過Kafka消息組可使同一條消息被不同Consumer消費，用戶離線和實時兩條線。

阿里巴巴開源項目: 基于mysql數(shù)據(jù)庫binlog的增量訂閱&消費 - agapple - ITeye技術網(wǎng)站
http://agapple.iteye.com/blog/1796633

背景
早期，阿里巴巴B2B公司因為存在杭州和美國雙機房部署，存在跨機房同步的業(yè)務需求。不過早期的數(shù)據(jù)庫同步業(yè)務，主要是基于trigger的方式獲取增量變更，不過從2010年開始，阿里系公司開始逐步的嘗試基于數(shù)據(jù)庫的日志解析，獲取增量變更進行同步，由此衍生出了增量訂閱&消費的業(yè)務，從此開啟了一段新紀元。ps. 目前內(nèi)部使用的同步，已經(jīng)支持mysql5.x和oracle部分版本的日志解析

基于日志增量訂閱&消費支持的業(yè)務：
數(shù)據(jù)庫鏡像
數(shù)據(jù)庫實時備份
多級索引 (賣家和買家各自分庫索引)
search build
業(yè)務cache刷新
價格變化等重要業(yè)務消息

項目介紹
名稱：canal [k?'n?l]
譯意： 水道/管道/溝渠
語言： 純java開發(fā)
定位： 基于數(shù)據(jù)庫增量日志解析，提供增量數(shù)據(jù)訂閱&消費，目前主要支持了mysql

工作原理
mysql主備復制實現(xiàn)

從上層來看，復制分成三步：
master將改變記錄到二進制日志(binary log)中（這些記錄叫做二進制日志事件，binary log events，可以通過show binlog events進行查看）；
slave將master的binary log events拷貝到它的中繼日志(relay log)；
slave重做中繼日志中的事件，將改變反映它自己的數(shù)據(jù)。

canal的工作原理：

原理相對比較簡單：
canal模擬mysql slave的交互協(xié)議，偽裝自己為mysql slave，向mysql master發(fā)送dump協(xié)議
mysql master收到dump請求，開始推送binary log給slave(也就是canal)
canal解析binary log對象(原始為byte流)

架構

說明：
server代表一個canal運行實例，對應于一個jvm
instance對應于一個數(shù)據(jù)隊列 （1個server對應1..n個instance)

instance模塊：
eventParser (數(shù)據(jù)源接入，模擬slave協(xié)議和master進行交互，協(xié)議解析)
eventSink (Parser和Store鏈接器，進行數(shù)據(jù)過濾，加工，分發(fā)的工作)
eventStore (數(shù)據(jù)存儲)
metaManager (增量訂閱&消費信息管理器)

知識科普
mysql的Binlay Log介紹
http://dev.mysql.com/doc/refman/5.5/en/binary-log.html
http://www.taobaodba.com/html/474_mysqls-binary-log_details.html

簡單點說：
mysql的binlog是多文件存儲，定位一個LogEvent需要通過binlog filename + binlog position，進行定位

收藏代碼

mysql> show variables like 'binlog_format';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| binlog_format | ROW |
+---------------+-------+
1 row in set (0.00 sec)

目前canal只能支持row模式的增量訂閱(statement只有sql，沒有數(shù)據(jù)，所以無法獲取原始的變更日志)

EventParser設計
大致過程：

點擊查看原始大小圖片

整個parser過程大致可分為幾步：
Connection獲取上一次解析成功的位置 (如果第一次啟動，則獲取初始指定的位置或者是當前數(shù)據(jù)庫的binlog位點)
Connection建立鏈接，發(fā)送BINLOG_DUMP指令 // 0. write command number // 1. write 4 bytes bin-log position to start at // 2. write 2 bytes bin-log flags // 3. write 4 bytes server id of the slave // 4. write bin-log file name
Mysql開始推送Binaly Log
接收到的Binaly Log的通過Binlog parser進行協(xié)議解析，補充一些特定信息// 補充字段名字，字段類型，主鍵信息，unsigned類型處理
傳遞給EventSink模塊進行數(shù)據(jù)存儲，是一個阻塞操作，直到存儲成功
存儲成功后，定時記錄Binaly Log位置

mysql的Binlay Log網(wǎng)絡協(xié)議：

說明：
圖中的協(xié)議4byte header，主要是描述整個binlog網(wǎng)絡包的length
binlog event structure，詳細信息請參考： http://dev.mysql.com/doc/internals/en/binary-log.html

EventSink設計

說明：
數(shù)據(jù)過濾：支持通配符的過濾模式，表名，字段內(nèi)容等
數(shù)據(jù)路由/分發(fā)：解決1:n (1個parser對應多個store的模式)
數(shù)據(jù)歸并：解決n:1 (多個parser對應1個store)
數(shù)據(jù)加工：在進入store之前進行額外的處理，比如join

數(shù)據(jù)1:n業(yè)務
為了合理的利用數(shù)據(jù)庫資源， 一般常見的業(yè)務都是按照schema進行隔離，然后在mysql上層或者dao這一層面上，進行一個數(shù)據(jù)源路由，屏蔽數(shù)據(jù)庫物理位置對開發(fā)的影響，阿里系主要是通過cobar/tddl來解決數(shù)據(jù)源路由問題。
所以，一般一個數(shù)據(jù)庫實例上，會部署多個schema，每個schema會有由1個或者多個業(yè)務方關注

數(shù)據(jù)n:1業(yè)務
同樣，當一個業(yè)務的數(shù)據(jù)規(guī)模達到一定的量級后，必然會涉及到水平拆分和垂直拆分的問題，針對這些拆分的數(shù)據(jù)需要處理時，就需要鏈接多個store進行處理，消費的位點就會變成多份，而且數(shù)據(jù)消費的進度無法得到盡可能有序的保證。
所以，在一定業(yè)務場景下，需要將拆分后的增量數(shù)據(jù)進行歸并處理，比如按照時間戳/全局id進行排序歸并.

EventStore設計

1. 目前僅實現(xiàn)了Memory內(nèi)存模式，后續(xù)計劃增加本地file存儲，mixed混合模式
2. 借鑒了Disruptor的RingBuffer的實現(xiàn)思路

RingBuffer設計：

定義了3個cursor
Put : Sink模塊進行數(shù)據(jù)存儲的最后一次寫入位置
Get : 數(shù)據(jù)訂閱獲取的最后一次提取位置
Ack : 數(shù)據(jù)消費成功的最后一次消費位置

實現(xiàn)說明：
Put/Get/Ack cursor用于遞增，采用long型存儲
buffer的get操作，通過取余或者與操作。(與操作： cusor & (size - 1) , size需要為2的指數(shù)，效率比較高)

Instance設計

instance代表了一個實際運行的數(shù)據(jù)隊列，包括了EventPaser,EventSink,EventStore等組件。
抽象了CanalInstanceGenerator，主要是考慮配置的管理方式：
manager方式： 和你自己的內(nèi)部web console/manager系統(tǒng)進行對接。(目前主要是公司內(nèi)部使用)
spring方式：基于spring xml + properties進行定義，構建spring配置.

Server設計

server代表了一個canal的運行實例，為了方便組件化使用，特意抽象了Embeded(嵌入式) / Netty(網(wǎng)絡訪問)的兩種實現(xiàn)
Embeded : 對latency和可用性都有比較高的要求，自己又能hold住分布式的相關技術(比如failover)
Netty : 基于netty封裝了一層網(wǎng)絡協(xié)議，由canal server保證其可用性，采用的pull模型，當然latency會稍微打點折扣，不過這個也視情況而定。(阿里系的notify和metaq，典型的push/pull模型，目前也逐步的在向pull模型靠攏，push在數(shù)據(jù)量大的時候會有一些問題)

增量訂閱/消費設計

canal的get/ack/rollback協(xié)議和常規(guī)的jms協(xié)議有所不同，允許get/ack異步處理，比如可以連續(xù)調(diào)用get多次，后續(xù)異步按順序提交ack/rollback，項目中稱之為流式api.
流式api設計的好處：
get/ack異步化，減少因ack帶來的網(wǎng)絡延遲和操作成本 (99%的狀態(tài)都是處于正常狀態(tài)，異常的rollback屬于個別情況，沒必要為個別的case犧牲整個性能)
get獲取數(shù)據(jù)后，業(yè)務消費存在瓶頸或者需要多進程/多線程消費時，可以不停的輪詢get數(shù)據(jù)，不停的往后發(fā)送任務，提高并行化. (作者在實際業(yè)務中的一個case：業(yè)務數(shù)據(jù)消費需要跨中美網(wǎng)絡，所以一次操作基本在200ms以上，為了減少延遲，所以需要實施并行化)

流式api設計：

每次get操作都會在meta中產(chǎn)生一個mark，mark標記會遞增，保證運行過程中mark的唯一性
每次的get操作，都會在上一次的mark操作記錄的cursor繼續(xù)往后取，如果mark不存在，則在last ack cursor繼續(xù)往后取
進行ack時，需要按照mark的順序進行數(shù)序ack，不能跳躍ack. ack會刪除當前的mark標記，并將對應的mark位置更新為last ack cusor
一旦出現(xiàn)異常情況，客戶端可發(fā)起rollback情況，重新置位：刪除所有的mark, 清理get請求位置，下次請求會從last ack cursor繼續(xù)往后取

數(shù)據(jù)對象格式：EntryProtocol.proto
Java代碼

收藏代碼

Entry
Header
logfileName [binlog文件名]
logfileOffset [binlog position]
executeTime [發(fā)生的變更]
schemaName
tableName
eventType [insert/update/delete類型]
entryType [事務頭BEGIN/事務尾END/數(shù)據(jù)ROWDATA]
storeValue [byte數(shù)據(jù),可展開，對應的類型為RowChange]

RowChange
isDdl [是否是ddl變更操作，比如create table/drop table]
sql [具體的ddl sql]
rowDatas [具體insert/update/delete的變更數(shù)據(jù)，可為多條，1個binlog event事件可對應多條變更，比如批處理]
beforeColumns [Column類型的數(shù)組]
afterColumns [Column類型的數(shù)組]

Column
index
sqlType [jdbc type]
name [column name]
isKey [是否為主鍵]
updated [是否發(fā)生過變更]
isNull [值是否為null]
value [具體的內(nèi)容，注意為文本]

說明：
可以提供數(shù)據(jù)庫變更前和變更后的字段內(nèi)容，針對binlog中沒有的name,isKey等信息進行補全
可以提供ddl的變更語句

HA機制設計
canal的ha分為兩部分，canal server和canal client分別有對應的ha實現(xiàn)
canal server: 為了減少對mysql dump的請求，不同server上的instance要求同一時間只能有一個處于running，其他的處于standby狀態(tài).
canal client: 為了保證有序性，一份instance同一時間只能由一個canal client進行get/ack/rollback操作，否則客戶端接收無法保證有序。

整個HA機制的控制主要是依賴了zookeeper的幾個特性，watcher和EPHEMERAL節(jié)點(和session生命周期綁定)，可以看下我之前zookeeper的相關文章。

Canal Server:

大致步驟：
canal server要啟動某個canal instance時都先向zookeeper進行一次嘗試啟動判斷 (實現(xiàn)：創(chuàng)建EPHEMERAL節(jié)點，誰創(chuàng)建成功就允許誰啟動)
創(chuàng)建zookeeper節(jié)點成功后，對應的canal server就啟動對應的canal instance，沒有創(chuàng)建成功的canal instance就會處于standby狀態(tài)
一旦zookeeper發(fā)現(xiàn)canal server A創(chuàng)建的節(jié)點消失后，立即通知其他的canal server再次進行步驟1的操作，重新選出一個canal server啟動instance.
canal client每次進行connect時，會首先向zookeeper詢問當前是誰啟動了canal instance，然后和其建立鏈接，一旦鏈接不可用，會重新嘗試connect.

Canal Client的方式和canal server方式類似，也是利用zokeeper的搶占EPHEMERAL節(jié)點的方式進行控制.

最后
項目的代碼： https://github.com/alibabatech/canal
這里給出了如何快速啟動Canal Server和Canal Client的例子，如有問題可隨時聯(lián)系
Quick Start
http://agapple.iteye.com/blog/1796070
https://github.com/alibabatech/canal/wiki/QuickStart

Client Example
http://agapple.iteye.com/blog/1796620
https://github.com/alibabatech/canal/wiki/ClientExample