承接上篇文章 ElasticSearch Rest/RPC 接口解析,這篇文章我們重點(diǎn)分析讓ES步入數(shù)據(jù)分析領(lǐng)域的Aggregation相關(guān)的功能和設(shè)計(jì)。
前言
我記得有一次到一家公司做內(nèi)部分享，然后有研發(fā)問我，即席分析這塊，他們用ES遇到一些問題。我當(dāng)時(shí)直接就否了，我說ES還是個(gè)全文檢索引擎，如果要做分析，還是應(yīng)該用Impala,Phenix等這種主打分析的產(chǎn)品。隨著ES的發(fā)展，我現(xiàn)在對(duì)它的看法，也有了比較大的變化。而且我認(rèn)為ES+Spark SQL組合可以很好的增強(qiáng)即席分析能夠處理的數(shù)據(jù)規(guī)模，并且能夠?qū)崿F(xiàn)復(fù)雜的邏輯，獲得較好的易用性。
需要說明的是，我對(duì)這塊現(xiàn)階段的理解也還是比較淺。問題肯定有不少，歡迎指正。
Aggregations的基礎(chǔ)
Lucene 有三個(gè)比較核心的概念：
- 倒排索引
- fieldData/docValue
- Collector
倒排索引不用我講了，就是term -> doclist的映射。
fieldData/docValue 你可以簡(jiǎn)單理解為列式存儲(chǔ),索引文件的所有文檔的某個(gè)字段會(huì)被單獨(dú)存儲(chǔ)起來。 對(duì)于這塊，Lucene 經(jīng)歷了兩階段的發(fā)展。第一階段是fieldData ，查詢時(shí)從倒排索引反向構(gòu)成doc-term。這里面有兩個(gè)問題：
- 數(shù)據(jù)需要全部加載到內(nèi)存
- 第一次構(gòu)建會(huì)很慢
這兩個(gè)問題其實(shí)會(huì)衍生出很多問題：最嚴(yán)重的自然是內(nèi)存問題。所以lucene后面搞了DocValue，在構(gòu)建索引的時(shí)候就生成這個(gè)文件。DocValue可以充分利用操作系統(tǒng)的緩存功能，如果操作系統(tǒng)cache住了，則速度和內(nèi)存訪問是一樣的。
另外就是Collector的概念，ES的各個(gè)Aggregator 實(shí)現(xiàn)都是基于Collector做的。我覺得你可以簡(jiǎn)單的理解為一個(gè)迭代器就好，所有的候選集都會(huì)調(diào)用Collector.collect(doc)方法，這里collect == iterate 可能會(huì)更容易理解些。
ES 能把聚合做快，得益于這兩個(gè)數(shù)據(jù)結(jié)構(gòu)，一個(gè)迭代器。我們大部分聚合功能，其實(shí)都是在fieldData/docValue 上工作的。
Aggregations 分類
Aggregations種類分為:
- Metrics
- Bucket
Metrics 是簡(jiǎn)單的對(duì)過濾出來的數(shù)據(jù)集進(jìn)行avg,max等操作，是一個(gè)單一的數(shù)值。
Bucket 你則可以理解為將過濾出來的數(shù)據(jù)集按條件分成多個(gè)小數(shù)據(jù)集，然后Metrics會(huì)分別作用在這些小數(shù)據(jù)集上。
對(duì)于最后聚合出來的結(jié)果，其實(shí)我們還希望能進(jìn)一步做處理，所以有了Pipline Aggregations,其實(shí)就是組合一堆的Aggregations 對(duì)已經(jīng)聚合出來的結(jié)果再做處理。
Aggregations 類設(shè)計(jì)
下面是一個(gè)聚合的例子：
{
"aggregations": {
"user": {
"terms": {
"field": "user",
"size": 10,
"order": {
"_count": "desc"
}
}
}
}
}
其語義類似這個(gè)sql 語句： select count(*) as user_count group by user order by user_count desc。
對(duì)于Aggregations 的解析，基本是順著下面的路徑分析：
TermsParser ->
TermsAggregatorFactory ->
GlobalOrdinalsStringTermsAggregator
在實(shí)際的一次query里，要做如下幾個(gè)階段：
Query Phase 此時(shí) 會(huì)調(diào)用GlobalOrdinalsStringTermsAggregator的Collector 根據(jù)user 的不同進(jìn)行計(jì)數(shù)。
RescorePhase
SuggestPhase
AggregationPhase 在該階段會(huì)會(huì)執(zhí)行實(shí)際的aggregation build,
aggregator.buildAggregation(0)，也就是一個(gè)特定Shard(分片)的聚合結(jié)果MergePhase。這一步是由接受到請(qǐng)求的ES來完成，具體負(fù)責(zé)執(zhí)行Merge(Reduce)操作
SearchPhaseController.merge。這一步因?yàn)闀?huì)從不同的分片拿到數(shù)據(jù)再做Reduce,也是一個(gè)內(nèi)存消耗點(diǎn)。所以很多人會(huì)專門搞出幾臺(tái)ES來做這個(gè)工作，其實(shí)就是ES的client模式，不存數(shù)據(jù)，只做接口響應(yīng)。
在這里我們我們可以抽取出幾個(gè)比較核心的概念：
- AggregatorFactory (生成對(duì)應(yīng)的Aggregator)
- Aggregation (聚合的結(jié)果輸出)
- Aggregator (聚合邏輯實(shí)現(xiàn))
另外值得注意的，PipeLine Aggregator 我前面提到了，其實(shí)是對(duì)已經(jīng)生成的Aggregations重新做加工，這個(gè)工作是只能單機(jī)完成的，會(huì)放在請(qǐng)求的接收端執(zhí)行。
Aggregation Bucket的實(shí)現(xiàn)
前面的例子提到，在Query 階段，其實(shí)就會(huì)調(diào)用Aggregator 的collect 方法，對(duì)所有符合查詢條件的文檔集都會(huì)計(jì)算一遍,這里我們涉及到幾個(gè)對(duì)象：
- doc id
- field (docValue)
- IntArray 對(duì)象
collect 過程中會(huì)得到 doc id,然后拿著docId 到 docValue里去拿到field的值(一般而言字符串也會(huì)被編碼成Int類型的)，然后放到IntArray 進(jìn)行計(jì)數(shù)。如果多個(gè)doc id 在某filed里的字段是相同的，則會(huì)遞增計(jì)數(shù)。這樣就實(shí)現(xiàn)了group by 的功能了。
Spark-SQL 和 ES 的組合
我之前一直在想這個(gè)問題，后面看了下es-hadoop的文檔，發(fā)現(xiàn)自己有些思路和現(xiàn)在es-hadoop的實(shí)現(xiàn)不謀而合。主要有幾點(diǎn)：
- Spark-SQL 的 where 語句全部(或者部分)下沉到 ES里進(jìn)行執(zhí)行，依賴于倒排索引，DocValues，以及分片,并行化執(zhí)行，ES能夠獲得比Spark-SQL更優(yōu)秀的響應(yīng)時(shí)間
- 其他部分包括分片數(shù)據(jù)Merge(Reduce操作，Spark 可以獲得更好的性能和分布式能力)，更復(fù)雜的業(yè)務(wù)邏輯都交給Spark-SQL (此時(shí)數(shù)據(jù)規(guī)模已經(jīng)小非常多了)，并且可以做各種自定義擴(kuò)展，通過udf等函數(shù)
- ES 無需實(shí)現(xiàn)Merge操作,可以減輕內(nèi)存負(fù)擔(dān)，提升并行Merge的效率(并且現(xiàn)階段似乎ES的Reduce是只能在單個(gè)實(shí)例里完成)
