大數(shù)據(jù)下的BI新特性

大數(shù)據(jù)BI的新需求包括大量化(多個大數(shù)據(jù)集并行分析)、多樣化(結(jié)構(gòu)化、半結(jié)構(gòu)化、非結(jié)構(gòu)化）、快速化(Velocity)和價值(易用性)。而計算分層（流計算、塊計算、全局計算）、快速分析（冗余維度、數(shù)據(jù)常駐在內(nèi)存中分析）和接近價值（業(yè)務(wù)人員易用的命令、靈活的編程

今天，主要介紹的是Twitter剛剛開源的一個計算框架，結(jié)合在一起實現(xiàn)一個快速靈活的架構(gòu)。先介紹一下我們Admaster的BI業(yè)務(wù)，我們對BI的定義是什么呢？統(tǒng)計，發(fā)現(xiàn)，預測。Admaster數(shù)據(jù)主要來自于門戶、客戶端、硬廣。門戶比如搜狐，新浪，客戶端，比如QQ迅雷。我們數(shù)據(jù)量非常大，基本上幾個TB，還有采集，我們有自己的搜集在線問卷，同時還有微博論壇，所以我們數(shù)據(jù)其實是有結(jié)構(gòu)化，有非結(jié)構(gòu)化的。
    Admaster系統(tǒng)不適用于我們，左邊這兩個圖可以看到，這些機柜非常昂貴，我們處于成本考慮不會使用這些東西。右上角HPCC Systems，是一個更好的系統(tǒng)，為什么不使用？因為生態(tài)系統(tǒng)非常封閉，沒有辦法把接口和我們自己的程序結(jié)合起來。其次，他開源出來的社區(qū)版比企業(yè)版性能差很多。所以，我們不會使用這樣的機器。
    我們看一下傳統(tǒng)分子方法，很多互聯(lián)網(wǎng)公司都是這樣分子數(shù)據(jù)。原始日志采集Scribe到Hadoop，然后到Hive結(jié)構(gòu)理念，在做OLAP分析，比如多維數(shù)據(jù)查詢。Hadoop是最近幾年才出來的，但是整套BI分析流程，其實還是《數(shù)據(jù)倉庫》這本書里面的關(guān)系模型，沒有超出這個框架的。
    互聯(lián)網(wǎng)數(shù)據(jù)最常見的一個分析例子是用戶行為分析，我們采集用戶訪問信息，用戶看到什么廣告，點擊了什么廣告，購買了什么，去了哪里。還有一些非結(jié)構(gòu)化的信息，半結(jié)構(gòu)化的信息，比如他發(fā)表的什么評論，微博上寫的什么東西，還有圖象，音頻，視頻，一些非結(jié)構(gòu)化的數(shù)據(jù)。可見我們數(shù)據(jù)特點是有以下幾個大到必須分布式存儲，因為我們基本上都超過TB每天，而且有一個趨勢不斷在增長。今年可能是TB每天，明年可能就是2TB，4TB，事實上我們今年數(shù)據(jù)比去年翻了5倍，我們需要一個非常平滑的分布式架構(gòu)，只需要加機器就可以，不需要改我的分析算法。
    還有海量數(shù)據(jù)集非常多，有很多海量級數(shù)據(jù)集進行分析。因為我們采集全部都是日式數(shù)據(jù)，所以數(shù)據(jù)量非常少。還有我們要把結(jié)構(gòu)化，半結(jié)構(gòu)化的數(shù)據(jù)合并起來進行分析。因為我們數(shù)據(jù)集擁有這些特點，所以特點決定了存儲和分析的方式。首先只有append，沒有更新和刪除，這種方式和剛才OceanBase是相似的，從字面上理解，首先要結(jié)果一個讀的操作，然后再去進行一個寫的措施，首先更新和刪除開銷肯定比讀和寫要大。
    其次，如果我們不做更新和刪除，實質(zhì)上是規(guī)避了一致性。我們只有更新數(shù)據(jù)相當于帶著時間和全新的數(shù)據(jù)，其實就不可能發(fā)生一致性的問題，什么最終一致性，因果一致性等等完全就規(guī)避掉了。其實今天上午大家講的這些，大家也可以看出來一致性是非常頭疼的問題。但是我們BI分析應用，完全規(guī)避了一致性問題。
    其次我們要盡可能的反觀新模式，反關(guān)系模式里面有一個雪花模型和星型模型，特別大的數(shù)據(jù)量情況下，我們要盡可能把緯度表冗余到事實表里面，盡量不做交易，特別一些小交易要放到事實表里面。還有一個特點，我們對計算框架靈活性的要求越來越好，而且很多算法更好理解，表達力更強。但是，就像今天上午巨先生講的一樣，MapReduce表現(xiàn)力還是受限，特別我們要一些發(fā)達算法的情況下有很大的問題。比如我在MapReduce使用一個共享數(shù)據(jù)群，首先編程非常復雜，其次大小是受限的，我不能靈活使用一個共享數(shù)據(jù)集。我們現(xiàn)在其實漸漸轉(zhuǎn)到DRPC，本質(zhì)就是一個分布式的遠程調(diào)用，凡是在單機上能實現(xiàn)的分機算法一定可以利用DRPC實現(xiàn)，我們可以利用其實現(xiàn)各種各樣的靈活分析算法。
    這是BI對我們提出的新需求，首先多個非常大的數(shù)據(jù)集放在一起分析，然后我們要把結(jié)構(gòu)化，半結(jié)構(gòu)化，非結(jié)構(gòu)化的數(shù)據(jù)放在一起分析，這就決定了不可能用基于關(guān)系型，結(jié)構(gòu)化型的數(shù)據(jù)集，把結(jié)構(gòu)化，半結(jié)構(gòu)化，非結(jié)構(gòu)化混合在一起進行分析，還有要快，Hadoop本質(zhì)是一個離線分析，首先啟動非常慢，從硬盤讀取數(shù)據(jù)，我們需要一個非常快，甚至達到秒級，讓用戶無縫體現(xiàn)的一個平臺，其次我們還需要業(yè)務(wù)人員盡可能簡單的使用它。
    為了解決剛才的這些需求，現(xiàn)在有以下一些BI方案。首先是計算分層，我們把數(shù)據(jù)分析的算法按照計算方式，分成三層，流計算層，塊計算，全局計算，一會我會一個層一個層介紹?？焖俜治鲇袃煞N方案，一種是冗余緯度，其實應該叫做冗余緯度的索引，一會會詳細介紹，還有一個方案把數(shù)據(jù)進行分析常駐在，使用內(nèi)存分析系統(tǒng)。還有使用非常靈活的框架，非常靈活的數(shù)據(jù)挖掘算法。
    這是我們使用的一個完整數(shù)據(jù)架構(gòu)，從采集數(shù)據(jù)進入流計算層，從流計算層進入快數(shù)據(jù)和全計算層，每個齒輪其實代表著數(shù)百臺服務(wù)器，整個架構(gòu)是一個分布式的。流數(shù)據(jù)，會把觸發(fā)的事件寫入關(guān)系型數(shù)據(jù)庫中，會把持續(xù)計算的結(jié)果寫到里面，快數(shù)據(jù)層會把大的數(shù)據(jù)進行分析結(jié)果寫到里面，同時全局數(shù)據(jù)，一般是把離線結(jié)果也寫進里面，對于結(jié)果和關(guān)系數(shù)據(jù)庫做一個對比，通過緩存層提供給用戶。流數(shù)據(jù)框架，塊數(shù)據(jù)框架，全局數(shù)據(jù)框架我們使用一種分析語言，本質(zhì)上相當于編程一次可以在三個地方同時使用，這樣的話無論是對開發(fā)人員，還是對業(yè)務(wù)人員來說都非常簡單。
    首先講一下計算分層，比如我們可以簡單的把數(shù)據(jù)分成實施層和P處理層，實施層把數(shù)據(jù)常駐在內(nèi)存當中可以達到非?？斓拿爰壏治觯看尾樵?，我們同時去實施層和P處理層查詢，然后把兩個結(jié)果合并起來，這樣才能得到我們最終結(jié)果。因為分層本質(zhì)其實一般來說，都是按時間來分的，比如數(shù)據(jù)量非常大，每天有一個TB進來，我們只能把今天TB數(shù)據(jù)放在里面，把今天之前的數(shù)據(jù)全部進行離線分析。今天數(shù)據(jù)我可以做到一個實時分析，我在調(diào)用，比如昨天半夜跑出來的結(jié)果，把那個結(jié)果加起來，就可以得到一個非常快，相當于進一步實施分析的結(jié)果。
    我們看一下Twitter的例子，是可以做到實時分析的。怎么做？Hadoop批量分析結(jié)果可能半夜做，寫到他自己開發(fā)的一個KV系統(tǒng)里面，批量導入的數(shù)據(jù)非?？欤赜诳焖俚膶?，把這個結(jié)果加起來反饋給客戶。
    我們介紹一下Twitter開源的流計算框架，用途有以下幾種。首先他可以做一個最快的ETL工具，ETL搞BI的都比較清楚，一個是抽取，轉(zhuǎn)換，轉(zhuǎn)入，大家可以簡單理解數(shù)據(jù)格式轉(zhuǎn)換，就把數(shù)據(jù)格式變成我們所需要的。然后他還可以利用冗余緯度數(shù)據(jù)，把小的緯度數(shù)據(jù)盡可能冗余到事實表里面，Storm是非常適合做這個事情。還有事件驅(qū)動報警，什么是事件驅(qū)動報警呢？舉個例子，比如我們監(jiān)測門戶上面廣告網(wǎng)站，我們監(jiān)測一條一個廣告發(fā)生了一個點擊，但是他之前10分鐘我們并沒有發(fā)現(xiàn)有暴光，  這是不合常理的，一個廣告怎么可能被點擊了卻沒有被人看到了，這可能是一個作弊事件，我們就需要記錄到數(shù)據(jù)庫里面。
    還有一個例子，我們在5分鐘之內(nèi)發(fā)現(xiàn)同一個IP下面發(fā)現(xiàn)1千次點擊，很明顯這是一個機器控制的作弊程序，也記錄到數(shù)據(jù)庫里面，說明只是一個作弊事件。還有Storm可以實現(xiàn)一種計算，叫做持續(xù)計算，請注意持續(xù)計算是一種受限算法，能夠持續(xù)計算的計算是“易并行”的，什么意思呢？比如典型的“易并行”問題，就是求coint，我可以分到10臺服務(wù)器上，每一臺服務(wù)器專門求一個行數(shù)，反饋給我加起來，這個結(jié)果肯定正確。同時，我上5分鐘得出求和結(jié)果，和下5分鐘求和結(jié)果加起來依然是正確的。
    不適用的問題是什么呢？比如數(shù)據(jù)去重統(tǒng)計，count，distinct，只能對全局數(shù)據(jù)去重。流計算，能夠計算的問題都是一些非常簡單的“易并行問題，其實在很多場合也比較有用，比如記數(shù)。Storm，吞吐性能非常好，因為底層使用了ZeroMQ，ZeroMQ性能非常好，比傳統(tǒng)要好的很多，達到了每秒將近2多萬消息吞吐量，還是在一臺PC上面，要超過百萬很清楚，本質(zhì)上就是內(nèi)存拷貝，沒有任何的持久化。顯而易見他是有安全性的問題，如果服務(wù)器掉鏈，正在處理的消息就完全消失了，那怎么辦呢？
    我們用Storm可以確保每行消息都被正確處理，失敗消息打回去重新處理。我們可以看右邊這個圖，比如一條消息是一句話，做什么事情，對每個單詞進行記錄，比如他把這句話六個單詞分還進行處理，最后一個失敗怎么了辦？每個任務(wù)都會發(fā)送一個ACK給Storm，Storm收到5個ACK，他就認為這個消息處理全部失敗，所以全部打回重新處理。通過這樣的方式來確保整個Stosm不會失效，出錯。
    同時還有最后一道防線，用P處理層來校正，整個Storm完蛋了，Twitter也發(fā)生類似事情，Storm失效之后，當然他是有人工錯誤引起的，并不是Storm本身的問題，就可以從，比如過了12小時之后，可以把從這當中把數(shù)據(jù)導出到Storm里面。Storm處理數(shù)據(jù)格式是基于元組tuple，類似MapReduce數(shù)據(jù)處理序列，本質(zhì)是DRPC編程框架，最靈活的進行編程。但是，我們需要自己控制內(nèi)存，Storm不是所有事情都做到了。
    一句話性肉元組在很多機器內(nèi)存中進行M和R不斷變化著自己的結(jié)構(gòu)，最后變成我們想要的元組形式?？梢钥匆粋€例子，最左邊正方題里面是所有交易數(shù)據(jù)，A代表A商品，客戶可能賣了三個型號，又一個客戶買了B35商品，我們按照客戶購買列出來，把哪種商品最熱賣的型號列出來，其實可以看最右邊的結(jié)果，我們希望生成這樣的結(jié)果，客戶購買一次商品是B和C，是B的35和A的23，客戶購買商品是A，最終賣的是5，我們中間怎么做呢？我們只需要執(zhí)行三個任務(wù)，比如第一個任務(wù)調(diào)整數(shù)據(jù)格式，其實相當于把原數(shù)據(jù)發(fā)送到三臺服務(wù)器上，把后面省略號去掉。然后我們在把數(shù)據(jù)相當于按照商品A，B，C做一個覆蓋，發(fā)送兩臺服務(wù)器上，大家可以看到最上面相當于按照A分類匯總，得出A結(jié)果，同時我們統(tǒng)計一下A的次序，生成A括號5，逗號2。同時再按頻次來做，最后得到我們想要的結(jié)果，整個流程非常簡單。
    大家還可以注意到一點，任務(wù)1，任務(wù)2，任務(wù)3，并不是分的很清楚，他只是把自己叫做精簡的任務(wù)處理，他只是在這些任務(wù)之間到底要不做grrouping，任務(wù)1和任務(wù)2，任務(wù)2和任務(wù)3之間是按照某個字段進行一個對比。我們可以看出Storm流計算就是一個Bolt接龍，數(shù)據(jù)相當于有一個噴射器，從最左邊一行行發(fā)射出來，之后通過整個相當于任務(wù)A進行分布式處理，每個圓圈代表一個服務(wù)器，最后得出我們想要的結(jié)果。同時，他在任務(wù)處理的過程中還可以輕易進行交互，我們BoltA，B，C，D，E，組成一個相當于類似Hadoop概念，會一直常駐在內(nèi)存當中，不管什么時候發(fā)生數(shù)據(jù)，我一直常駐在內(nèi)存中，來這樣數(shù)據(jù)就進行這樣的處理邏輯。所以，他是一個持續(xù)性的流計算框架。
    來多少數(shù)據(jù)我處理多少，我可以變成來一條處理一條，我也可以來5千條處理5千條得出一個結(jié)果。所以，他是一個不停在運行的流式計算框架。我們可以看一下代碼的例子，非常簡單。我們可以先建一Storm，然后我們在Storm加一個數(shù)據(jù)噴射口，不斷噴出單詞，我們做第一個任務(wù)，隨機接受數(shù)據(jù)噴射口傳出來數(shù)據(jù)，我們可以做一個格式轉(zhuǎn)化，我還可以指定數(shù)據(jù)用幾臺服務(wù)器來處理，比如逗號3是第一個，我們用3臺服務(wù)器處理。最后一行，可能是用fieldsGruping，相當于做一個操作，非常簡單。
    但是，這個例子其實會錯誤運行的，數(shù)據(jù)源源不斷發(fā)射，最后會源源不斷會導致超出服務(wù)器內(nèi)存，服務(wù)器內(nèi)存會爆掉。實際上我們需要每割一段時間就去清理Bolt里面的內(nèi)存，把里面內(nèi)存釋放掉，需要我自己手工控制內(nèi)存，這是一個難點。我們講一下快計算，為什么需要快計算，顯然是實施之后更有價值。比如下面這個圖可以發(fā)現(xiàn)，在早上6點鐘的時候，新西蘭某個汽車品牌發(fā)現(xiàn)銷量上漲，我們就立刻捕捉這個趨勢，就做出一些決策。實際上洞察的決策，動態(tài)決策非常有價值，特別是現(xiàn)在，舉個例子我們互聯(lián)網(wǎng)廣告行業(yè)，在美國不能說所有廣告，相當一部分廣告份額已經(jīng)將近三分之一了，是從實時廣告計價平臺上購買的，包括雅虎，微軟也有，相當于一條在線廣告實時發(fā)布出去，我們各個采購商去競價，如果我這個采購商，或者說廣告代理商能夠在秒級發(fā)現(xiàn)一個數(shù)據(jù)，能夠在秒級發(fā)現(xiàn)價格有這樣的上漲，或者某個地區(qū)，某個廣告可以實現(xiàn)非常好的CTS，我可以立刻搶拍下這個廣告收入增加，這時這在國外是非?；罨鸬膭討B(tài)決策，盡可能快的進行挖掘發(fā)現(xiàn)趨勢。
    我們?nèi)绻麑崿F(xiàn)快速的塊計算，在塊上做呢？淘寶有一個很好的prom方案，甚至可以做到毫秒級響應，當然有一些缺點，只能固定分析幾個緯度，同時導入數(shù)據(jù)花時間，相當于把所有冗余索引犧牲內(nèi)存孔，現(xiàn)在淘寶在很多產(chǎn)品里面使用prom方案，他可以把前一天所有支付寶交易數(shù)據(jù)，按照不同的緯度導入不同Server里面。大家可以看里面這個圖，Data Server相當于上午型和個人用途，筆記本電腦索引，有商務(wù)型索引寫在Server里面，個人用電腦寫在另一個Server里面。
    然后Data Server2都寫在里面，Data Server品牌寫在里面，基于D品牌筆記本成交總量，我們把13寸所有都拿過來，得到5和6兩條索引，我們再去某臺實際Data Server上得到的交易金額，加起來返回給客戶，通過這樣一種索引達到非常快速的查詢。所以，哪怕是幾千萬數(shù)據(jù)，上億數(shù)據(jù)，我們依然可以做到在秒級之內(nèi)返回。大家可以看到這一頁，每個索引都要增加一臺服務(wù)器，如果緯度固定還好，如果有上千臺緯度是不是需要準備上千個服務(wù)器，肯定就沒有辦法用這個方案了。我們現(xiàn)在在用一個Storm+內(nèi)存文件系統(tǒng)，在內(nèi)存文件系統(tǒng)上調(diào)用Storm有一個模塊，相當于LinearDRPCT opologyBuilder，包括消息處理，失敗處理都分裝好了。
    優(yōu)點是什么算法非常靈活，我可以實現(xiàn)各種各樣的數(shù)據(jù)分析算法，而且足夠的快。一般來說，只要數(shù)據(jù)，比如只有幾個G基本上都是秒級可以算出來，比如我?guī)装倥_服務(wù)器，每臺64G內(nèi)存，我可以實現(xiàn)幾個TB數(shù)據(jù)存儲，全局全部內(nèi)存之中。當然缺點也非常明顯，大家也都可以看的出來，數(shù)據(jù)非常容易失掉，如果掉鏈了全部數(shù)據(jù)都沒有，我需要從硬盤上再倒出來，非?；〞r間。同時，我能夠分析熱數(shù)據(jù)大小，其實完全受限，我不能只分配給內(nèi)存系統(tǒng)，其實不停在服務(wù)器終端大幅度移動，和Hadoop有本質(zhì)不同，Hadoop設(shè)計思想是轉(zhuǎn)移計算到數(shù)據(jù)這邊，絕對不是讓數(shù)據(jù)在不同服務(wù)器之間移動，那是絕對不行的，因為Hadoop這么快，設(shè)計思想就是把計算移到數(shù)據(jù)這邊。
    我在內(nèi)存之間移動數(shù)據(jù)這個開銷還是可以接受的，因為硬盤拷貝還是足夠快的。其實我們現(xiàn)在最院士K-means聚類的算法是這樣的，比如這么一個閃點圖，我要找到比較密集的幾個圖，比如隨機指示三個點，對每一個點進行一個聚類計算，我認為我是這個中心點里面的人，這樣每個點都可以歸納到一個里面，重新求一個中心點，這樣重新計算中心點，實際越來越近，補貼的迭代，這個位置會越來越向中心點移動。
    如果我用K-means重新計算，都會把這些數(shù)據(jù)搭到硬盤上面來移動非常慢，如果我們有Storm的話，很簡單，數(shù)據(jù)永遠在這之中，這個算法會比K-means上快幾百倍，非常正常。我們回顧一下流計算，塊計算，全局計算。Storm同時承擔了ETL，流計算，塊計算工作，流計算就是一些易并行簡單算法，還可以報警，發(fā)生事件組合報警，還有反觀模式，同時還可以對內(nèi)存文件系統(tǒng)上的數(shù)據(jù)進行一個快速計算快速迭代。Hadoop很明顯做的和Storm不太一樣，是廉價全局大數(shù)據(jù)計算。我剛才講的很多算法必須在全局數(shù)據(jù)上計算，比如一個最簡單的驅(qū)動數(shù)據(jù)，如果能夠忍受誤差可以這么算，有一定錯誤質(zhì)量算法，如果你要得到一個非常準確的認同，必須在Hadoop上進行運轉(zhuǎn)。所以，這兩者之間完全不可能替代，Hadoop絕對做不到Storm那么快。Storm部分算法也實現(xiàn)不了，只能在Hadoop中做一個全局問題，需要其進行補充，所以兩者之間非常好。
    現(xiàn)在我們整個比較框架，就是Storm和Hadoop，兩者互相配合，互相補充，能夠?qū)崟r計算盡量放到Storm上去做，Hadoop是必須的一個廉價全局大計算增長。因為我們整個BI架構(gòu)是拼在一起的，相當于現(xiàn)在有兩套系統(tǒng)，我們開發(fā)了DSL，這是針對業(yè)務(wù)定制語言。首先Hadoop Cascalog太復雜，首先要有一些可以簡單的邏輯代碼，看一看右下角這是Hadoop只有五行代碼就實現(xiàn)了，非常簡單。所以說，作為DSL可以在Admaster上看一下，甚至MapReduce都發(fā)現(xiàn)有四種DSL，我們自己還開發(fā)針對業(yè)務(wù)DSL，對業(yè)務(wù)人員不需要知道業(yè)務(wù)是在Hadoop上面還是storm上面，也無需知道數(shù)據(jù)是結(jié)構(gòu)化還是半結(jié)構(gòu)化。
    左下角是我們開發(fā)的SUM，類似于微軟MDS，實現(xiàn)一個分析功能，我列出13寸，14寸，15村，日期在2011-11-20之前，寫出A品牌，我把這個交易額數(shù)據(jù)進行一個分類匯總得出結(jié)果。實際實現(xiàn)邏輯是非常簡單的，有一個命令解析器得到緯度和度量及，數(shù)據(jù)解析，信息傳遞到Hadoop的map的reduce和storm的bolt中，篩選數(shù)據(jù)拼接維度。工作流拼接，根據(jù)功能拼接，Hadoop job和storm topology，合并查詢，組合storm和Hadoop的查詢結(jié)果。最后這些結(jié)果寫入到里面，類似于剛才Twitter的一種做法，同時去查詢storm和Hadoop分析結(jié)果，得到最終一個匯總。
    我們再回顧一下完整的BI分析架構(gòu)，這個圖和剛才略有差別。這是采集數(shù)據(jù)進來之后進到storm，數(shù)據(jù)報警寫在Hadoop里面，把采集到數(shù)據(jù)storm變形，一部分寫入Hadoop之中，一部分寫入內(nèi)存分件系統(tǒng)，我們自己開發(fā)一套BSL，同時在HadoopDSL上跑得到分析結(jié)果。當然一般來說，offline跑昨天的數(shù)據(jù)，實時分析結(jié)果顯示在Mongodb里面，有一些持續(xù)計算也寫進Mongodb而里面，一個查詢需要三個地方加起來，得到一個精確實時的結(jié)果。我可以得到，比如一秒前的統(tǒng)計結(jié)果，比如一個廣告上線之后，就可以知道這個廣告在1秒之前被暴光多少次，我從Hadoop，我從內(nèi)存文件系統(tǒng)，我從Storm等結(jié)果加起來就可以知道一秒之前發(fā)生的任何事情。
    最后結(jié)果將被寫到MongoDB，然后用戶可以進行查詢得到結(jié)果。用MongoDB的原因，我們需要對三個層次最后的分析結(jié)果，我們需要在做一些分析，MongoDB的分析還是很強的，有時候我們需要在對統(tǒng)計之后的結(jié)果進行再統(tǒng)計，這是MongoDB的想象。同時，我們也有非常好的MongoDB專家，所以我們可以保證能夠吃透MongoDB。