分享人介紹：王團(tuán)結(jié)，七牛數(shù)據(jù)平臺(tái)工程師，主要負(fù)責(zé)數(shù)據(jù)平臺(tái)的設(shè)計(jì)研發(fā)工作。關(guān)注大數(shù)據(jù)處理，高性能系統(tǒng)服務(wù)，關(guān)注Hadoop、Flume、Kafka、Spark等離線、分布式計(jì)算技術(shù)。
下為討論實(shí)錄
數(shù)據(jù)平臺(tái)在大部分公司屬于支撐性平臺(tái)，做的不好立刻會(huì)被吐槽，這點(diǎn)和運(yùn)維部門很像。所以在技術(shù)選型上優(yōu)先考慮現(xiàn)成的工具，快速出成果，沒必要去擔(dān)心有技術(shù)負(fù)擔(dān)。早期，我們走過彎路，認(rèn)為沒多少工作量，收集存儲(chǔ)和計(jì)算都自己研發(fā)，發(fā)現(xiàn)是吃力不討好。去年上半年開始，我們?nèi)鎿肀ч_源工具，搭建自己的數(shù)據(jù)平臺(tái)。
數(shù)據(jù)平臺(tái)設(shè)計(jì)架構(gòu)
公司的主要數(shù)據(jù)來源是散落在各個(gè)業(yè)務(wù)服務(wù)器上的半結(jié)構(gòu)化的日志(系統(tǒng)日志、程序日志、訪問日志、審計(jì)日志等)。大家有沒考慮過為什么需要日志？日志是最原始的數(shù)據(jù)記錄，如果不是日志，肯定會(huì)有信息上的丟失。說個(gè)簡(jiǎn)單的例子，需求是統(tǒng)計(jì)nginx上每個(gè)域名的的流量，這個(gè)完全可以通過一個(gè)簡(jiǎn)單的nginx模塊去完成，但是當(dāng)我們需要統(tǒng)計(jì)不同來源的流量時(shí)就法做了。所以需要原始的完整的日志。
有種手法是業(yè)務(wù)程序把日志通過網(wǎng)絡(luò)直接發(fā)送出去，這并不可取，因?yàn)榫W(wǎng)絡(luò)和接收端并不完全可靠，當(dāng)出問題時(shí)會(huì)對(duì)業(yè)務(wù)造成影響或者日志丟失。對(duì)業(yè)務(wù)侵入最小最自然的方式是把日志落到本地硬盤上。
Agent設(shè)計(jì)需求
每臺(tái)機(jī)器上會(huì)有一個(gè)agent去同步這些日志，這是個(gè)典型的隊(duì)列模型，業(yè)務(wù)進(jìn)程在不斷的push，agent在不停的pop。agent需要有記憶功能，用來保存同步的位置(offset)，這樣才盡可能保證數(shù)據(jù)準(zhǔn)確性，但不可能做到完全準(zhǔn)確。由于發(fā)送數(shù)據(jù)和保存offset是兩個(gè)動(dòng)作，不具有事務(wù)性，不可避免的會(huì)出現(xiàn)數(shù)據(jù)不一致性情況，通常是發(fā)送成功后保存offset，那么在agent異常退出或機(jī)器斷電時(shí)可能會(huì)造成多余的數(shù)據(jù)。
agent需要足夠輕，這主要體現(xiàn)在運(yùn)維和邏輯兩個(gè)方面。agent在每臺(tái)機(jī)器上都會(huì)部署，運(yùn)維成本、接入成本是需要考慮的。agent不應(yīng)該有解析日志、過濾、統(tǒng)計(jì)等動(dòng)作，這些邏輯應(yīng)該給數(shù)據(jù)消費(fèi)者。倘若agent有較多的邏輯，那它是不可完成的，不可避免的經(jīng)常會(huì)有升級(jí)變更動(dòng)作。
數(shù)據(jù)收集流程
數(shù)據(jù)收集這塊的技術(shù)選擇，agent 是用go自己研發(fā)的，消息中間件kafka，數(shù)據(jù)傳輸工具flume。說到數(shù)據(jù)收集經(jīng)常有人拿flume和kafka做比較，我看來這兩者定位是不同的，flume更傾向于數(shù)據(jù)傳輸本身，kakfa是典型的消息中間件用于解耦生產(chǎn)者消費(fèi)者。
具體架構(gòu)上，agent并沒把數(shù)據(jù)直接發(fā)送到kafka，在kafka前面有層由flume構(gòu)成的forward。這樣做有兩個(gè)原因
1. kafka的api對(duì)非jvm系的語言支持很不友好，forward對(duì)外提供更加通用的http接口
2. forward層可以做路由、kafka topic和kafka partition key等邏輯，進(jìn)一步減少agent端的邏輯
forward層不含狀態(tài)，完全可以做到水平擴(kuò)展，不用擔(dān)心成為瓶頸。出于高可用考慮，forward通常不止一個(gè)實(shí)例，這會(huì)帶來日志順序問題，agent 按一定規(guī)則(round-robin、failover等)來選擇forward實(shí)例，即使kafka partition key一樣，由于forward層的存在，最終落入kafka的數(shù)據(jù)順序和 agent發(fā)送的順序可能會(huì)不一樣。我們對(duì)亂序是容忍的，因?yàn)楫a(chǎn)生日志的業(yè)務(wù)基本是分布式的，保證單臺(tái)機(jī)器的日志順序意義不大。如果業(yè)務(wù)對(duì)順序性有要求，那得把數(shù)據(jù)直接發(fā)到kafka，并選擇好partition key，kafka只能保證 partition級(jí)的順序性。
跨機(jī)房收集要點(diǎn)
多機(jī)房的情形，通過上述流程，先把數(shù)據(jù)匯到本地機(jī)房kafka 集群，然后匯聚到核心機(jī)房的kafka，最終供消費(fèi)者使用。由于kafka的mirror對(duì)網(wǎng)絡(luò)不友好，這里我們選擇更加的簡(jiǎn)單的flume去完成跨機(jī)房的數(shù)據(jù)傳送。
flume在不同的數(shù)據(jù)源傳輸數(shù)據(jù)還是比較靈活的，但有幾個(gè)點(diǎn)需要注意
1. memory-channel效率高但可能有丟數(shù)據(jù)的風(fēng)險(xiǎn)，file-channel安全性高但性能不高。我們是用memory-channel，但把capacity設(shè)置的足夠小，使內(nèi)存中的數(shù)據(jù)盡可能少，在意外重啟和斷電時(shí)丟的數(shù)據(jù)很少。個(gè)人比較排斥file-channel，效率是一方面，另一個(gè)是對(duì)flume的期望是數(shù)據(jù)傳輸，引入file-channel時(shí)，它的角色會(huì)向存儲(chǔ)轉(zhuǎn)變，這在整個(gè)流程中是不合適的。通常flume的sink端是kafka和hdfs這種可用性和擴(kuò)張性比較好的系統(tǒng)，不用擔(dān)心數(shù)據(jù)擁堵問題。
2. 默認(rèn)的http souce 沒有設(shè)置線程池，有性能問題，如果有用到，需要自己修改代碼。
3. 單sink速度跟不上時(shí)，需要多個(gè)sink。像跨機(jī)房數(shù)據(jù)傳輸網(wǎng)絡(luò)延遲高單rpc sink吞吐上不去和hdfs sink效率不高情形，我們?cè)谝粋€(gè)channel后會(huì)配十多個(gè)sink。
Kafka使用要點(diǎn)
kafka在性能和擴(kuò)展性很不錯(cuò)，以下幾個(gè)點(diǎn)需要注意下
1. topic的劃分，大topic對(duì)生產(chǎn)者有利且維護(hù)成本低，小topic對(duì)消費(fèi)者比較友好。如果是完全不相關(guān)的相關(guān)數(shù)據(jù)源且topic數(shù)不是發(fā)散的，優(yōu)先考慮分topic。
2. kafka的并行單位是partition，partition數(shù)目直接關(guān)系整體的吞吐量，但parition數(shù)并不是越大越高，3個(gè)partition就能吃滿一塊普通硬盤io了。所以partition數(shù)是由數(shù)據(jù)規(guī)模決定，最終還是需要硬盤來抗。
3. partition key選擇不當(dāng)，可能會(huì)造成數(shù)據(jù)傾斜。在對(duì)數(shù)據(jù)有順序性要求才需使用partition key。kafka的producer sdk在沒指定partition key時(shí)，在一定時(shí)間內(nèi)只會(huì)往一個(gè)partition寫數(shù)據(jù)，這種情況下當(dāng)producer數(shù)少于partition數(shù)也會(huì)造成數(shù)據(jù)傾斜，可以提高producer數(shù)目來解決這個(gè)問題。
數(shù)據(jù)到kafka后，一路數(shù)據(jù)同步到hdfs，用于離線統(tǒng)計(jì)。另一路用于實(shí)時(shí)計(jì)算。由于今天時(shí)間有限，接下來只能和大家分享下實(shí)時(shí)計(jì)算的一些經(jīng)驗(yàn)
實(shí)時(shí)計(jì)算我們選擇的spark streaming。我們目前只有統(tǒng)計(jì)需求，沒迭代計(jì)算的需求，所以spark streaming使用比較保守，從kakfa讀數(shù)據(jù)統(tǒng)計(jì)完落入mongo中，中間狀態(tài)數(shù)據(jù)很少。帶來的好處是系統(tǒng)吞吐量很大，但幾乎沒遇到內(nèi)存相關(guān)問題
spark streaming對(duì)存儲(chǔ)計(jì)算結(jié)果的db tps要求較高。比如有10w個(gè)域名需要統(tǒng)計(jì)流量，batch interval為10s，每個(gè)域名有4個(gè)相關(guān)統(tǒng)計(jì)項(xiàng)，算下來平均是4w tps，考慮到峰值可能更高，固態(tài)硬盤上的mongo也只能抗1w tps，后續(xù)我們會(huì)考慮用redis來抗這么高的tps
有外部狀態(tài)的task邏輯上不可重入的，當(dāng)開啟speculation參數(shù)時(shí)候，可能會(huì)造成計(jì)算的結(jié)果不準(zhǔn)確。說個(gè)簡(jiǎn)單的例子
這是個(gè)把計(jì)算結(jié)果存入mongo的task
這個(gè)任務(wù)，如果被重做了，會(huì)造成落入mongo的結(jié)果比實(shí)際多。
有狀態(tài)的對(duì)象生命周期不好管理，這種對(duì)象不可能做到每個(gè)task都去new一個(gè)。我們的策略是一個(gè)jvm內(nèi)一個(gè)對(duì)象，同時(shí)在代碼層面做好并發(fā)控制。類似下面。
在spark 1.3的后版本，引入了 kafka direct api試圖來解決數(shù)據(jù)準(zhǔn)確性問題，使用direct在一定程序能緩解準(zhǔn)確性問題，但不可避免還會(huì)有一致性問題。為什么這樣說呢？direct api 把kafka consumer offset的管理暴露出來(以前是異步存入zookeeper)，當(dāng)保存計(jì)算結(jié)果和保存offset在一個(gè)事務(wù)里，才能保證準(zhǔn)確。
這個(gè)事務(wù)有兩種手段做到，一是用mysql這種支持事務(wù)的數(shù)據(jù)庫保存計(jì)算結(jié)果offset，一是自己實(shí)現(xiàn)兩階段提交。這兩種方法在流式計(jì)算里實(shí)現(xiàn)的成本都很大。
其次direct api 還有性能問題，因?yàn)樗接?jì)算的時(shí)候才實(shí)際從kafka讀數(shù)據(jù)，這對(duì)整體吞吐有很大影響。
要分享的就這些了，最后秀下我們線上的規(guī)模。flume ＋ kafka ＋ spark 8臺(tái)高配機(jī)器，日均500億條數(shù)據(jù)，峰值 80w tps。
