大數據下的BI新特性

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

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