基于zookeeper和storm的車載流式計(jì)算框架

1、基于zookeeper和storm的車載流式計(jì)算框架摘要n 案例背景n 日本第一車載SAAS平臺(tái)n 案例需求n 數(shù)據(jù)量劇增，吞吐量劇增n 車載機(jī)數(shù)量劇增+數(shù)據(jù)庫(kù)連接數(shù)劇增n 任務(wù)分布不均導(dǎo)致實(shí)時(shí)性無(wú)法滿足n 案例技術(shù)方案n 案例技術(shù)方案概覽n 數(shù)據(jù)存儲(chǔ)平臺(tái)MongoDBn 實(shí)時(shí)計(jì)算平臺(tái)Stormn 集群協(xié)調(diào)ZooKeepern 最佳實(shí)踐n MongoDB介紹及最佳實(shí)踐n Storm介紹及最佳實(shí)踐n ZooKeeper介紹及最佳實(shí)踐n 架構(gòu)不足及經(jīng)驗(yàn)分享案例背景1業(yè)務(wù)背景n 業(yè)務(wù)場(chǎng)景 物流公司 校車安全 油氣公司 n 主要功能 運(yùn)行數(shù)據(jù)記錄 監(jiān)視駕駛 違規(guī)警告 績(jī)效考評(píng) 報(bào)表輸出 危險(xiǎn)地帶 實(shí)時(shí)

2、監(jiān)控日本市場(chǎng)占有率第一，積極拓展國(guó)內(nèi)業(yè)務(wù)WebAPP中間件層硬件層通信層分析計(jì)算層1分析計(jì)算層2Web ServerWriteWriteWriteReadWritePollingReadRMDBPrimitiveLog基本Bs_WorkSheetBs_TimeChartBs_FreqDataBs_DigiData集計(jì)Master設(shè)定動(dòng)態(tài)管理RMDBPrimitiveLog基本Bs_WorkSheetBs_TimeChartBs_FreqDataBs_DigiData集計(jì)Master設(shè)定動(dòng)態(tài)管理RMDBPrimitiveLog基本Bs_WorkSheetBs_TimeChartBs_FreqDa

3、taBs_DigiData集計(jì)Master設(shè)定動(dòng)態(tài)管理案例背景2技術(shù)背景市場(chǎng)競(jìng)爭(zhēng)激烈，維持第一的保證PAAS一 數(shù)據(jù)量劇增+吞吐量劇增2013年車輛數(shù)：10000臺(tái)數(shù)據(jù)量：3000w/天吞吐量：80M/S ()1000200040005000700010000050001000015000200220042006200820102012車輛數(shù)車輛數(shù)車両數(shù)：10000Commn ServerAnalysisServerStasticServer分時(shí)日月年bs_primitiveLogWRW-62,3403,722,15729,763,261654,722,397 7,856,6448,198bs

4、_WorkSheet-RWRW8,422504,8754,032,87488,704,7821,064,448,769bs_TimeChart-RWRW10,873600,8744,804,430105,600,1001,267,200,296bs_FreqData-WR45027,0964216,7846,480,54077,760,054bs_DegitachoData-W-1,01060,079480,10010,560,006126,720,190數(shù)據(jù)特性Heavy Write Heavy Read Past Useless2015年車輛數(shù)：100000臺(tái)數(shù)據(jù)量：30000w/天吞吐量

5、：800M/S：車載機(jī)發(fā)送數(shù)據(jù)最高頻率2次/秒，每次發(fā)送4KB數(shù)據(jù)。10000臺(tái)車載機(jī)的峰值為10000*2*4KB =80000KB=80M/S ：從這個(gè)數(shù)據(jù)可以算出實(shí)際平均吞吐量是62340*4KB/60 = 4M/s, 但是固定時(shí)間點(diǎn)車載機(jī)會(huì)同時(shí)向云端發(fā)送運(yùn)行數(shù)據(jù)()此表抽出通信中間件分析集計(jì)中間件DBbs_primitiveLogbs_WorkSheetbs_TimeChartbs_FreqDatabs_DegitachoData設(shè)置信息寫原始日志表車載機(jī)數(shù)據(jù)二 連接數(shù)劇增+客戶劇增DB1DB2DB3DBn 通信服務(wù)器與各個(gè)業(yè)務(wù)DB建立長(zhǎng)連接 目前的SAAS平臺(tái)已經(jīng)有600租戶 每個(gè)通

6、信服務(wù)器要與600個(gè)DB建立數(shù)據(jù)庫(kù)連接歸一化迫在眉睫NoSQL10W車輛時(shí)：1.需要17臺(tái)通信服務(wù)器2.峰值：10W*8KB/S=800MB/S數(shù)據(jù)量長(zhǎng)尾來(lái)襲 現(xiàn)在有600個(gè)客戶公司，未來(lái)？三 分析計(jì)算程序任務(wù)不均勻+實(shí)時(shí)性不夠DB1DB2DB3DBnAnalysis ProcessAnalysis ProcessAnalysis ProcessAnalysis Processn 現(xiàn)狀 同一公司的車輛數(shù)據(jù)分布在同一數(shù)據(jù)庫(kù) 分析計(jì)算進(jìn)程與DB一對(duì)一地分析數(shù)據(jù)n 問(wèn)題 若一個(gè)公司有很多輛車，那么一個(gè)分析計(jì)算進(jìn)程應(yīng)付不過(guò)來(lái)，無(wú)法分散計(jì)算 若一個(gè)公司車輛很少，那么該公司對(duì)應(yīng)的分析進(jìn)程將沒(méi)有任務(wù)，浪費(fèi)服

7、務(wù)器資源 因?yàn)榉治鲞M(jìn)程的緩慢執(zhí)行，車輛的實(shí)時(shí)運(yùn)行數(shù)據(jù)無(wú)法反映到客戶端案例技術(shù)方案概覽案例技術(shù)方案概覽通信集群消息隊(duì)列+存儲(chǔ)計(jì)算集群數(shù)據(jù)庫(kù)集群協(xié)調(diào)集群應(yīng)用集群接受連接負(fù)載均衡通信直連事件網(wǎng)絡(luò)消息發(fā)布消息訂閱數(shù)據(jù)庫(kù)操作存儲(chǔ)消息負(fù)載均衡-數(shù)據(jù)分片主從備份負(fù)載均衡無(wú)模式數(shù)據(jù)分解數(shù)據(jù)分析數(shù)據(jù)統(tǒng)計(jì)日志分析數(shù)據(jù)存儲(chǔ)模式隔離私有倉(cāng)庫(kù)節(jié)點(diǎn)管理任務(wù)分配狀態(tài)維護(hù)Open APIBrowseripadPhoneLibeventKafkaMongoDBStormSQL ServerZooKeeperIIS、WCF數(shù)據(jù)存儲(chǔ)平臺(tái)MongoDBMongoDB是一個(gè)介于關(guān)系數(shù)據(jù)庫(kù)和非關(guān)系數(shù)據(jù)庫(kù)之間的產(chǎn)品，是非關(guān)系數(shù)據(jù)庫(kù)當(dāng)中功能

8、最豐富，最像關(guān)系數(shù)據(jù)庫(kù)的。他支持的數(shù)據(jù)結(jié)構(gòu)非常松散，是類似json的bjson格式，因此可以存儲(chǔ)比較復(fù)雜的數(shù)據(jù)類型。Mongo最大的特點(diǎn)是他支持的查詢語(yǔ)言非常強(qiáng)大，其語(yǔ)法有點(diǎn)類似于面向?qū)ο蟮牟樵冋Z(yǔ)言，幾乎可以實(shí)現(xiàn)類似關(guān)系數(shù)據(jù)庫(kù)單表查詢的絕大部分功能，而且還支持對(duì)數(shù)據(jù)建立索引。它的特點(diǎn)是高性能、易部署、易使用，存儲(chǔ)數(shù)據(jù)非常方便。主要功能特性有： 面向集合存儲(chǔ)，易存儲(chǔ)對(duì)象類型的數(shù)據(jù) 模式自由 支持動(dòng)態(tài)查詢 支持完全索引，包含內(nèi)部對(duì)象 支持查詢 支持復(fù)制和故障恢復(fù) 使用高效的二進(jìn)制數(shù)據(jù)存儲(chǔ)，包括大型對(duì)象（如視頻等） 自動(dòng)處理分片，以支持云計(jì)算層次的擴(kuò)展性 支持RUBY，PYTHON，JAVA，C+，

9、PHP等多種語(yǔ)言。 文件存儲(chǔ)格式為BSON（一種JSON的擴(kuò)展） 可通過(guò)網(wǎng)絡(luò)訪問(wèn)數(shù)據(jù)量劇增和連接數(shù)劇增的解決辦法n 高頻讀寫表抽出n 1公司：1DB模式N公司：1DBDB1DB2DB3DBnDB1DB1DB2DB3DBn選擇MongoDB的原因n 增刪改查操作最接近SQL，遷移容易，并可為其他業(yè)務(wù)使用n 支持?jǐn)?shù)據(jù)的自動(dòng)和手動(dòng)分片，橫向擴(kuò)展容易n 性能測(cè)試結(jié)果接近預(yù)期，夠用即可，不用追求完美n 支持?jǐn)?shù)據(jù)復(fù)制、故障轉(zhuǎn)移、橫向擴(kuò)展，基本符合RAS需求CAP定理性能關(guān)系不大2個(gè)mongos，3個(gè)分片，非安全插入：?jiǎn)为?dú)執(zhí)行時(shí)：寫線程10個(gè)并發(fā)時(shí)，每秒37000次寫入讀線程4個(gè)并發(fā)時(shí)，每秒43000次讀取

10、同時(shí)執(zhí)行時(shí)：寫線程10個(gè)并發(fā)時(shí)，每秒達(dá)25000次寫入讀線程4個(gè)并發(fā)時(shí)，每秒達(dá)36000次讀?。篊PU 3.2GHz MEM 16G 千兆網(wǎng)卡MongoDB最佳實(shí)踐n 邏輯部署和物理部署n 存儲(chǔ)結(jié)構(gòu)設(shè)計(jì)n 定期數(shù)據(jù)清除ShardsLoad BalanceConfig serverReplica setsmongosmongosmongosmongodmongodmongodmongodmongodmongodmongodmongodmongodmongodmongodmongodC 2mongodC1 mongodC3 mongodclientMongoDB的部署邏輯架構(gòu)client192.1

11、68.0.2192.168.0.3192.168.0.4192.168.0.5192.168.0.6Shard1(master) Shard2(master)Shard3(master)Shard1(slave)Shard1(arbiter)Shard2(slave) Shard3(slave)Shard1(slave)Shard3(slave)Shard2(arbiter)Shard3(arbiter) Shard1(arbiter)Shard2(arbiter)Shard2(slave)Shard3 (arbiter)Config server1Config server2mongosmo

12、ngosmongosShard1：192.168.0.2,192.168.0.5,192.168.0.4,192.168.0.3,192.168.0.6(arbiter)Shard2：192.168.0.3,192.168.0.5,192.168.0.2,192.168.0.4,192.168.0.6(arbiter)Shard3：192.168.0.4,192.168.0.5,192.168.0.3,192.168.0.2,192.168.0.6(arbiter)Config Server1：192.168.0.2Config Server2：192.168.0.3Mongos：192.16

13、8.0.4，192.168.0.5，192.168.0.6MongoDB部署物理架構(gòu)存儲(chǔ)數(shù)據(jù)結(jié)構(gòu)設(shè)計(jì) “v”: “DTS19982221” “r”: “t” : 2013-11-14T08:17:00.016Z “d” : ”xxxxxxxxxxxxxxxxxx” “o” : “yyyyyyyyyyyyyyyyyy” , “t” : 2013-11-14T08:19:38.342Z“ “d” : ”xxxxxxxxxxxxxxxxxx” “o” : “yyyyyyyyyyyyyyyyyy” “s”: “zzzzz” 名字段要短 db.vehicle.save(“v”:” DTS19982221

14、 ”, “r: “t: 201399988772, “d”:”xxxxxxxxxxxxxxxx”,”o”:”yyyyyyyyyyy” db.vehicle.find(“v”:” DTS19982221 ”, “r: $elemMatch: “t:$gte: 201309898876251728定期數(shù)據(jù)清除n 清理程序（MonDC）運(yùn)行在ZooKeeper集群上，自身無(wú)狀態(tài)n 清理程序（MonDC）監(jiān)視MongoDB內(nèi)存使用量n 清理程序（MonDC）檢測(cè)刪除過(guò)期數(shù)據(jù)實(shí)時(shí)計(jì)算平臺(tái)Storm與Hadoop類似，Storm為分布式實(shí)時(shí)計(jì)算提供了一組通用原語(yǔ)?？杀挥糜凇傲魈幚怼敝校瑢?shí)時(shí)處理消息并更新

15、數(shù)據(jù)庫(kù)。這是管理隊(duì)列及工作者集群的另一種方式。Storm也可被用于“連續(xù)計(jì)算”， 對(duì)數(shù)據(jù)流做連續(xù)查詢，在計(jì)算時(shí)就將結(jié)果以流的形式輸出給用戶。它還可被用于“分布式RPC”，以并行的方式運(yùn)行昂貴的運(yùn)算。Storm的主工程師Nathan Marz表示：Storm可以方便地在一個(gè)計(jì)算機(jī)集群中編寫與擴(kuò)展復(fù)雜的實(shí)時(shí)計(jì)算，Storm之于實(shí)時(shí)處理，就好比Hadoop之于批處理。Storm保證每個(gè)消息都會(huì)得到處理，而且它很快在一個(gè)小集群中，每秒可以處理數(shù)以百萬(wàn)計(jì)的消息。更棒的是你可以使用任意編程語(yǔ)言來(lái)做開發(fā)。Storm的主要特點(diǎn)如下： 簡(jiǎn)單的編程模型 可以使用各種編程語(yǔ)言 容錯(cuò)性 水平擴(kuò)展 可靠的消息處理 快速

16、Storm為什么號(hào)稱實(shí)時(shí)計(jì)算系統(tǒng)？n 計(jì)算基于內(nèi)存，較之磁盤有數(shù)量級(jí)之差n 使用消息隊(duì)列做數(shù)據(jù)傳輸，速度比數(shù)據(jù)庫(kù)快很多n 使用流式計(jì)算思想，數(shù)據(jù)可以源源不斷的被處理n 任何一個(gè)節(jié)點(diǎn)都可以將中間計(jì)算結(jié)果反饋給用戶為什么選擇Storm？n 無(wú)需維護(hù)消息隊(duì)列（Topic、LB、Broker），專注Producer&Consumern 滿足業(yè)務(wù)需求，適合業(yè)務(wù)場(chǎng)景n 社區(qū)活躍，版本更新快n 性能卓越，易擴(kuò)展Storm里的核心概念n Tuple&StreamTupleTupleStreamn Topology&Spout&BoltSpoutBoltBoltBoltBoltB

17、oltClusterStorm里的核心概念n Nimbus&Supervisor&Worker&TaskNimbusZooKeeperSupervisorZooKeeperZooKeeperSupervisorSupervisorSupervisorSupervisorWorkerTask:SpoutTask:BoltTask:Bolt提交代碼，分配任務(wù)，監(jiān)視Supervisor運(yùn)行業(yè)務(wù)拓?fù)渚€程進(jìn)程Storm里的核心概念n Stream GroupingSpoutBoltABoltBoltBBoltBoltCI know who am IIKnow:1whoamI:1I

18、knowI:1who:1am:1NOT PASS問(wèn)題：如何保證tuple按照予定的路徑發(fā)送到指定Bolt呢？Storm里的核心概念n Stream GroupingShuffle Grouping 隨機(jī)分組， 隨機(jī)派發(fā)stream里面的tuple.Fields Grouping 按字段分組，具有同樣字段的tuple會(huì)被分到相同的Task。All Grouping 廣播發(fā)送， 對(duì)于每一個(gè)tuple， 所有的Bolts都會(huì)收到。Global Grouping 全局分組， 這個(gè)tuple被分配到storm中的一個(gè)bolt的其中一個(gè)task。再具體一點(diǎn)就是分配給id值最低的那個(gè)task。Non Gro

19、uping 不分組， 這個(gè)分組的意思是說(shuō)stream不關(guān)心到底誰(shuí)會(huì)收到它的tuple。Direct Grouping 直接分組, 這是一種比較特別的分組方法，用這種分組意味著消息的發(fā)送者指定由消息接收者的哪個(gè)task處理這個(gè)消息。Storm里的核心概念n Stream GroupingBoltABoltBBoltABoltBBoltABoltBBoltABoltBAllFieldsShuffleGlobalStorm最佳實(shí)踐n 計(jì)算拓?fù)湓O(shè)計(jì)n 數(shù)據(jù)順序性保證n 基于統(tǒng)計(jì)的Worker和Task數(shù)值設(shè)置計(jì)算拓?fù)湓O(shè)計(jì)n 數(shù)據(jù)輸入 FetchSpout MongoDB中讀取車輛運(yùn)行數(shù)據(jù)n 分析處理

20、AnalysisBolt 解碼運(yùn)行數(shù)據(jù)，如經(jīng)緯度、違反等n 數(shù)據(jù)準(zhǔn)備 DataPreBolt 為集計(jì)所需要的數(shù)據(jù)作準(zhǔn)備，會(huì)與數(shù)據(jù)庫(kù)交互n 集計(jì)處理 StasticBolt 統(tǒng)計(jì)一段時(shí)間或一天的運(yùn)行數(shù)據(jù)n 數(shù)據(jù)庫(kù)寫 DBWBolt 所有涉及寫數(shù)據(jù)庫(kù)的操作由此BOLT完成計(jì)算拓?fù)湓O(shè)計(jì)MongoDBFetchSpoutAnalysisBoltAnalysisBoltAnalysisBoltDataPreBoltDataPreBoltDBWBoltDBWBoltCompany 1Company 2Company 3Company 4DBMLayerStasticBoltStasticBoltFetch

21、SpoutFetchSpoutAnalysisBoltAnalysisBoltSpoutDataCollBoltDataCollBoltStasticBoltStasticBoltFileds GroupingFileds GroupingFileds GroupingDTS0002, XXXXXXDTS0001, XXXXXXDTS0001, YYYYYDTS0002, YYYYYDTS0001, ZZZZZZZDTS0002, ZZZZZZZFields Grouping 按字段分組， 比如按VehicleId的模n結(jié)果來(lái)分組， 具有同樣VehicleId的tuple會(huì)被分到相同的Bolt

22、s， 而不同的userid則會(huì)被分配到不同的Bolts。計(jì)算拓?fù)湓O(shè)計(jì)順序性如何保證最大效率化基于統(tǒng)計(jì)的計(jì)算量均勻化最佳參數(shù)設(shè)置 通過(guò)反復(fù)的測(cè)試，得到一個(gè)接近能夠處理完數(shù)據(jù)的比例設(shè)置，這里是2：10. 按照一個(gè)最佳數(shù)值設(shè)置建議：Worker個(gè)數(shù)是機(jī)器的整數(shù)倍，Task是Worker的整數(shù)倍。我們有7臺(tái)機(jī)器作為Supervisor，所以Worker數(shù)設(shè)置為7，Task數(shù)值為14，于是得到FetchSpout個(gè)數(shù)為2，AnalysisBolt個(gè)數(shù)為12. 以上設(shè)置后，計(jì)算延時(shí)降低到1.3s左右Storm Cluster Number： 8Zookeeper簡(jiǎn)介 Zookeeper是Google的Ch

23、ubby一個(gè)開源實(shí)現(xiàn)，是高有效和可靠的協(xié)同工作系統(tǒng)。 Zookeeper能夠用來(lái)leader選舉，配置信息維護(hù)等，在一個(gè)分布式的環(huán)境中，需要一個(gè)Master實(shí)例或存儲(chǔ)一些配置信息，確保文件寫入的一致性等。 ZooKeeper是一個(gè)分布式的，開放源碼的分布式應(yīng)用程序協(xié)調(diào)服務(wù)，包含一個(gè)簡(jiǎn)單的原語(yǔ)集，是Hadoop和Hbase的重要組件。 目前提供Java和C的接口。n 誰(shuí)在用ZooKeepern ZooKeeper是什么lZooKeeper軟件維護(hù)的是一個(gè)樹形的數(shù)據(jù)結(jié)構(gòu)l從任一個(gè)ZooKeeper節(jié)點(diǎn)看，樹視圖都一樣/root /dataNode1 /dataNode2 /dataNode3 /d

24、ataNode4 /dataNode5 /subDataNode1 /subDataNode2 /subDataNode1lZooKeeper集群維護(hù)的是一個(gè)樹形的數(shù)據(jù)結(jié)構(gòu)/root /dataNode1 /dataNode2 /dataNode3 /dataNode4 /dataNode5 /subDataNode1 /subDataNode2 /subDataNode1/root /dataNode1 /dataNode2 /dataNode3 /dataNode4 /dataNode5 /subDataNode1 /subDataNode2 /subDataNode1/root /dat

25、aNode1 /dataNode2 /dataNode3 /dataNode4 /dataNode5 /subDataNode1 /subDataNode2 /subDataNode1/root /dataNode1 /dataNode2 /dataNode3 /dataNode4 /dataNode5 /subDataNode1 /subDataNode2 /subDataNode1/root /dataNode1 /dataNode2 /dataNode3 /dataNode4 /dataNode5 /subDataNode1 /subDataNode2 /subDataNode1寫操作l

26、ZooKeeper各節(jié)點(diǎn)之間的數(shù)據(jù)結(jié)構(gòu)是同步的/root /dataNode1 /dataNode2 /dataNode3 /dataNode4 /dataNode5 /subDataNode1Zookeeper核心概念數(shù)據(jù)結(jié)構(gòu)Zookeeper核心概念節(jié)點(diǎn)類型節(jié)點(diǎn)特點(diǎn)功能PERSISTENT持久化目錄節(jié)點(diǎn)，節(jié)點(diǎn)數(shù)據(jù)不會(huì)丟失保存配置文件，小量數(shù)據(jù)存儲(chǔ)PERSISTENT_SEQUENTIAL順序自動(dòng)編號(hào)目錄節(jié)點(diǎn)隊(duì)列，共享鎖等EPHEMERAL臨時(shí)目錄節(jié)點(diǎn)，客戶端斷開即消失作為客戶端連接狀態(tài)的監(jiān)控節(jié)點(diǎn)EPHEMERAL_SEQUENTIAL臨時(shí)自動(dòng)編號(hào)目錄節(jié)點(diǎn)同上，但可作為隊(duì)列/root /da

27、taNode1 /dataNode2 /dataNode3 /dataNode4 /dataNode5 /subDataNode1 /subDataNode2 /subDataNode1n 統(tǒng)一命名服務(wù)（Name Service）n 配置管理（Configuration Management）n 集群管理（Group Membership）n 共享鎖（Locks）n 隊(duì)列管理 （Queue Management）Zookeeper的應(yīng)用場(chǎng)景ZooKeeper的使用n 協(xié)調(diào)數(shù)據(jù)清理（ MonDC ）n 協(xié)調(diào)任務(wù)分配（TaskDistribute）n 協(xié)調(diào)Spout對(duì)MongoDB的數(shù)據(jù)訪問(wèn)n 集群節(jié)點(diǎn)監(jiān)控?cái)?shù)據(jù)清理程序（ MonDC ）開始MongoDB輪詢清理警戒是否/shardProcTime /shard1：2013000912 /shard2：2013000523 /shard3：2013000943 /shard4：2013000125 /shard5：2013000229mongostat任務(wù)分配程序（TaskDistribute） spoutTasks