Druid 驅動海量實時多維分析

時間 2019-11-17

原文原文鏈接

我今天分享的話題是Druid驅動海量數據實時多維分析。javascript

一、需求背景

首先我來談一下海量實時多維分析的需求背景，咱們廣告系統有DSP睿視系統和AD exchange等，前段時間品友的曹老師分享時提到DSP Ad exchange,若是對DSP不瞭解的同窗，能夠在百度百科上查一下「互聯網廣告DSP」，實時競價是DSP的核心，廣告主或者優化師須要動態調整出價優化使收益最大化。html

廣告主調整出價策略或者投放策略進行優化，想要儘快獲得實時的反饋，好比修改了地域定投，實時分地域的競得率和轉化率，甚至是分鐘粒度的。在咱們的DSP系統中咱們提供12種維度(不算多)的多維分析，用戶能夠任意組合下鑽查詢。java

最初的時候我並非採用Druid，而是採用Storm+Kafka+Redis實時數據處理，以及hive+mysql的離線處理的Lambda架構架構見下圖。mysql

實時數據通過Storm ETL，主要是將不一樣維度組合做爲key，計算metrics之後爲value存在redis中。離線數據定時將昨天的數據在Hive中通過一系列ETL，按照維度組合計算metrics之後將結果存儲在mysql中。git

採用redis做爲實時數據的存儲有兩個核心問題須要解決：web

redis不支持range scan，咱們須要在app層拼好全部的key，而後調用mget獲取，若是執行groupby查詢的話例如select area,pv from t where cast='a' group byarea的話，需在客戶端好全部的cast_area的key。那麼問題來了，我並不知道投放a在哪些城市產生過曝光，若是有地域定向還好，若是沒有的話須要拼接全部城市的key取redis中進行mget。哪這樣只是城市維度還好，全國也只有幾百個城市，但若是換作其餘基數很大的維度，哪性能開銷就比較大了。redis

若是作到真正的多維分析，須要窮舉1維，2維知道N維維度的全部組合，例若有A B C和D，那麼一維的組合有A BC D，二維的有 AB BC CD DA等組合，依次類推直到組合到N維。同時每種維度又有本身的基數，例如ABCD各自的基數都是100，最差的狀況，今ABCD這種組合就會有1億個key，但現實數據通常都是稀疏，不會出現這種的狀況。總之窮舉維度組合做爲key的方式存儲空間會很是大，特別是添加維度之後，有多是指數級的增加，很是不可控。sql

爲了規避這個問題，咱們對功能進行了降級，咱們提供某些維度組合的查詢，這樣就作到了簡單可控。但須要歷史和實時結合查詢的場景，問題就又來了，我須要將mysql的數據和redis的數據整合起來。只能在app端進行數據合併。就這樣這個系統愉快地高效運轉了一段時間。數據庫

4月份的時候我參加Qcon和Druid中國用戶組的Meetup，聽Druid項目發起人Fangjin Yang將Druid的昨天今天和明天，6年前他們在metamarkets創業公司，主要爲廣告公司提供多維數據分析，嘗試過不少方案，例如傳統的關係型數據mysql/postgresql,kv存儲的 Hbase，Hbase受限於rowkey設計，都不能很好地解決多維分析，隨後建立Druid。我相信不少公司都有相似的經歷。說了這麼多終於引入正題。數組

二、Druid簡介

Druid是一個實時處理時序數據的Olap數據庫，由於它的索引首先按照時間分片，查詢的時候也是按照時間線去路由索引。

下面介紹一下它的幾個特性：

亞秒響應的交互式查詢。支持較高併發，爲面向用戶的平臺提供Olap查詢(注意這是相比其餘OLAP的強大優點)。
支持實時導入，導入便可被查詢。支持高併發導入。
採用分佈式shared-nothing的架構，能夠擴展到PB級。
支持聚合函數，count和sum，以及使用javascript實現自定義UDF。
支持複雜的Aggregator，近似查詢的Aggregator例如HyperLoglog以及Yahoo開源的DataSketches。
支持Groupby，Select，Search查詢。
不支持大表之間的Join，但其lookup功能知足和維度表的Join 。

三、它是如何作到的呢？

列存儲，倒排索引，Roll UP，roaring 或者concisebitmap位圖索引以及高效的壓縮支撐了Druid的高效查詢。提到倒排索引，你們都會聯想到Lucene，Solr以及ElasticSearch。確實Elastic Search是Druid的強大競爭對手，有興趣的同窗請移步http://druid.io/docs/latest/comparisons/druid-vs-elasticsearch.html，該部分不是本次分享的重點，就不作展開討論。

Druid是由一組不一樣角色的節點，每種節點都是一個單獨的子系統負責不一樣的功能，這些節點組成一個系統協同工做，咱們來看一下Druid的總體架構見下圖：

這張圖上有Realtime，Broker，Historical等節點，以及灰色的表明數據導入的流向，紅色的表明數據查詢的流向。

Segment是Druid的索引。Druid首先按照時間分片，Segment Granularity表明Segment存儲了那些時間範圍的數據，這裏的 Segment Granularity能夠是分鐘小時天等時間粒度。

RealTime部分能夠是Indexing Service或者Realtime Node，它負責實時導入數據，生成Segment(索引),在Realtime中採用LSM-tree 的模型，他能夠採用push或者pull的方式獲取流式數據，數據首先添加到內存的增量索引中，內存增量索引採用SkipList，對寫友好並且可以高併發處理，查詢也相對來講高效。

當達到必定閾值後，這個閾值能夠內存增量中最大條數，或者累積到必定時間。採用異步線程將內存增量索引轉成倒排索引持久化到磁盤中，同時生成一個新的內存增量索引接收數據。

周而復始，當達到設定的Segment Granularity時，Druid會Merge那些Segment Granularity內全部持久化的索引成一個Segment，而且推送到Deep Storage。在推送到 Deep Storage以前，Realtime負責對這些Segment的查詢。

Historical是Druid的主力，他從"Deep Storage"中下載Segment，採用mmap的方式加載Segment，並負責來自broker對這些 Segment的查詢。在Zookeeper中爲每一個Historical建立了一個Load Queue來決定下載那些Segment，一旦加載成功，Historical在 Zookeeper中宣告這個Segment由它來負責。

Historical採用多線程的方式處理高併發，好比一個請求過來涉及多個Segment的查詢，那麼會爲每一個segment分配一個線程，併發地執行，而後彙總結果返回給broker。因爲Historical處理Groupby查詢時，須要大量的內存存放中間臨時結果，那麼內存管理就顯得相當重要，Druid採用一個簡單的內存管理池，每次申請都會返回固定大小的direct buffer，使用完畢之後還回池中。

前面提到Druid採用Mmap的方式加載Segment，若是索引的總大小小於內存大小，而且留有必定量的direct memory大小，那麼Druid變成了內存數據庫，瓶頸不在IO，而不是在CPU，索引的壓縮以及數據的聚合都很耗CPU。Historical採用Shared-Nothing架構，能夠無限擴展，提高內存大小以及總的CPU core數從而提高計算能力。

Broker相似分佈式搜索引擎中的元搜索引擎，他不負責任何Segment的查詢，他只是一個代理，從Zookeeper中獲取TimeLine，這個 TimeLine記錄了intervals->List(Server)的mapping關係，接收到Client的請求之後，按照時間段在TimeLine查找Segment分佈在那些 Server上。而後併發地去查詢而後彙總結果。前面提到在每一個realtime或historical內部也是相似這樣的架構。

爲了便於你們理解，下面我用一幅圖來展現整個流程。

Coordinator負責協調Segment的均衡加載，Coordinator從元數據存儲mysql中獲取那些還未被加載的Segment，根據當前全部 Historical的負載能力均衡地分配到其LoadQueue。默認採用2副本加載，也就是說把同一個Segment分配到兩個Historical節點上去，這樣設計是爲了保障高可用。

Coordinator同時會檢測historical的存活，一旦發現某個Historical當掉或者有新的Historical加入，就會進行自動均衡。

好了通過上面對Druid的總體介紹，我相信你們已經有一個初步印象。重點說明的是Druid採用LSM-tree的方式作到實時索引實時查詢，下一步我將講述支撐高效查詢的索引以及壓縮。

三、支撐高效查詢的索引以及壓縮

咱們在討論高效查詢時，那麼得首先定義一個標準，如何才能稱之爲快，Druid官方宣稱他們已經實如今一萬億行數據集上執行 query，響應時間小於1秒。這裏提到的一萬億行數據集我理解是Raw Data，而不是導入Druid之後存儲的數據集大小。

由於Druid在導入數據過程當中有Rollup處理，這個特性會減小數據集的大小，固然這只是個人理解並無和做者求證。言歸正傳，Druid是如何作到這麼快的呢。索引是支撐數據庫高效查詢的基石。如下面的一組數據舉例來演示整個索引以及查詢過程。

這也是Druid要求的數據，第一列是時間，是必選項，廣告和地域都是維度列，在Druid中維度都做爲字符串處理，展現數和點擊數爲Metric，通常是看做數值處理，能夠是 Long或者Double，其實也能夠複雜的metric例如HyperLogLog來處理UV，DataSketches的功能更加加強能夠實現交併集的處理，能夠用作訪客的留存或者回訪統計，此處不詳細展開。

Raw Data的話，每一個展現或者點擊事件，其數量都是1，Druid在數據導入階段會進行Rollup，將相同維度組合的數據進行聚合處理，例如在2016/05/14T11期間，在廣告C1地域P1有14條件曝光事件，3條點擊事件，Rollup會將採用其設定的聚合器進行聚合，在這裏咱們採用sum的方式，獲得展現數14，點擊數3.

Druid採用倒排索引和列存儲的方式，列存儲採用字典編碼的方式，將某一個維度的值，按照字典順序編碼成整數，以廣告維度爲例，這個數據集中共有4條記錄，不重複的值有3個C1 、C2和C3，這裏提出一個基數的概念，基數能夠理解爲集合中不重複值的數量。那麼廣告維度的基數是3.咱們把它們按照字典順序編碼獲得C1=0，C2=1，C3=2。字典一般採用數組結構表達，元素的下標爲字典編碼值。

前面咱們將了Druid的數據格式，Rollup和列存儲以及字典編碼方式。

倒排索引和數據佈局

倒排索引被應用在搜索引擎上用於文檔的檢索，隨着大數據的發展，倒排索引一樣在大數據分析領域綻開出新的光芒，例如Elastic Search，愈來愈多的用戶把它用在數據分析上。

Druid一樣採用倒排索引，但不象Lucene，Druid專一於數據統計分析，並且只支持布爾查詢。Druid的索引結構佈局由字典，正排（列存儲）以及倒排組成，其中倒排的PostingList採用壓縮的BitMap位圖索引，BitMap的優點是能支持快速的布爾查詢，後續會介紹。

目前支持Consice和Roaring兩種BitMap方式，它們在稀疏的Posting List上均能起到很好的壓縮效果，更重要的是在進行布爾運算時不用解壓縮，直接在壓縮的結構上進行布爾運算。下圖展現的結構是將廣告編製成索引的數據佈局。例如C1只出如今第一行，那麼它的BitMap是1000，同理C2出如今第二和三行，那麼它的Bitmap是0110.這裏是介紹Dimension的索引布局，Metric一樣也採用列存儲的方式按塊壓縮。

說到索引，不得不提是壓縮技術。

前面提到BitMap支持Consice和Roaring兩種壓縮編碼方式，在Column正排部分使用變長整數編碼，不一樣於其餘場景下的可能須有一個bit標示分割，在Druid的索引中，字典的大小是已知的。若是字典的大小小於0Xf，那麼只有一個 byte存儲，同理，若是字典的大小小於0Xff則用兩個byte存儲，依次類推。若是維度的基數爲100，那個只有一個字節存儲，壓縮率爲25%，很是高效。

在最新的索引版本中在變長整數編碼的技術又加上了LZ系列壓縮。壓縮是一把雙刃劍，平衡的是IO和CPU的開銷。說到LZ系列壓縮，Druid採用按塊壓縮，默認塊大小爲64K。

布爾查詢，以這個SQL查詢爲例，selectsum(click) from table where time between 2016-05-24T11 and2016-05-24T12 and廣告=C2 and 地域=P1;首先根據時間段定位到Segment，而後根據廣告=C2和地域=P1，獲得他們各自的字典編碼，C2=1，P1=0，而後根據字典編碼獲得BitMap的Offet，從而獲得Bitmap，C2=1 index爲1的bitmap爲0110，同理獲得P1的bitmap爲1100，0110和1100進行And與運算是0100，獲得的offset是1，說明咱們須要在click中讀取offset=1的值是4.

再舉一個複雜一點的例子。

Group Byquery。Select 地域，sum(click) fromtable where time between 2016-05-24T11 and 2016-05-24T12 and 廣告=C2 group by 地域。Group By較上面的Select查詢不一樣的是它須要按照地域去彙總點擊數。這樣就得把符合過濾條件「廣告=C2」的地域和點擊數兩列的值都遍歷出來再在內存中作分組聚合。

接下來演示一下整個過程，首先是根據"廣告=C2"獲得字典編碼=1，而後根據index=1獲得bitmap是 0110，Offset是1和2，根據Offset分別在地域和點擊數中遍歷獲得，P1，4和P2 20.固然這個例子的數據很是少，現實中海量數據的話 Group by有可能遍歷大量的數據，首先在內存中作聚合再獲得最終的結果集。

在這一部分，咱們講了Druid高效查詢的基石，倒排索引，巧妙的索引結構佈局和高效的壓縮，BitMap以及布爾查詢。

四、應用案例

咱們回到應用背景中提到的案例，使用Druid之後的架構變成下圖。

特別要提出的是最初咱們採用RealtimeNode，部署簡單，但每次Schema的更新都要重啓每一個實時節點，而且實時節點沒有副本不能保障高可用，Indexing service做爲其的升級替代品，是一套獨立的分佈式系統能保障高可用，做者推薦使用。這個架構中，依然有離線和實時兩條線，離線會在T+1的時候修正數據。

目前咱們天天的數據在20億左右，13個維度，20多個指標項，大部分的查詢響應時間都在幾百毫秒，其導入便可查詢的實時性和高效的交互式查詢很好地支撐了業務需求。