ES學習筆記之-AvgAggregation的實現過程分析
我們需要檢視資料的統計量時,均值是最重要的特徵之一。
對於海量資料,這類簡單的聚合ES可以做到秒級別返回。聚合是ES的特色功能。
那麼ES是如何實現這一功能的呢?
我們知道,ES的資料儲存在各個節點中, 所以ES的實現AvgAggregation時基本思路就是先統計各個節點,然後彙總。
先了解ES是如何統計單個節點: 參考AvgAggregator
@Override public LeafBucketCollector getLeafCollector(LeafReaderContext ctx, final LeafBucketCollector sub) throws IOException { if (valuesSource == null) { return LeafBucketCollector.NO_OP_COLLECTOR; } final BigArrays bigArrays = context.bigArrays(); final SortedNumericDoubleValues values = valuesSource.doubleValues(ctx); return new LeafBucketCollectorBase(sub, values) { @Override public void collect(int doc, long bucket) throws IOException { counts = bigArrays.grow(counts, bucket + 1); sums = bigArrays.grow(sums, bucket + 1); values.setDocument(doc); final int valueCount = values.count(); counts.increment(bucket, valueCount); double sum = 0; for (int i = 0; i < valueCount; i++) { sum += values.valueAt(i); } sums.increment(bucket, sum); } }; }
即實現Collector類的collect()方法。然後通過doc_values
機制獲取文件相關欄位的值,分別匯入counts和sums兩個變數中。
收集完成counts和sums過後,就需要彙總各個節點的值, 這在搜尋的第二階段。 整個鏈路如下:
s1: 前端請求傳送到叢集某一節點的TransportSearchAction.doExecute()
方法中。
switch(searchRequest.searchType()) { ..... case QUERY_THEN_FETCH: searchAsyncAction = new SearchQueryThenFetchAsyncAction(logger, searchService, clusterService, indexNameExpressionResolver, searchPhaseController, threadPool, searchRequest, listener); break; ...... } searchAsyncAction.start();
見到start()方法,我以為這個是另啟一個執行緒,後面發現原來不是的。 這個start()方法把整個查詢過程分為兩個階段:
階段一:
performFirstPhase(), 即把請求分發到各個節點,然後記錄節點處理的結果。如果返回的分片是最後一個分片,則轉入階段二。
階段二:
performFirstPhase() -> onFirstPhaseResult() -> innerMoveToSecondPhase() -> moveToSecondPhase() 。這裡利用了模板設計模式。在階段二中,會再次向各個節點發起請求,通過docId獲取文件內容。
s2: 對於聚合而言, 階段二最重要的鏈路是moveToSecondPhase() -> executeFetch() -> finishHim() -> searchPhaseController.merge() , merge()中包含了如下的業務邏輯: 合併hits, 合併suggest, 合併addAggregation 等。 這裡我們關注聚合。
聚合的入口方法是InternalAggregations.reduce()
, 如果熟悉hadoop, reduce方法的執行邏輯看這個名字也能理解一部分。reduce的中文翻譯“歸納”,挺生動形象的。整個鏈路的入口為InternalAvg.doReduce()
。
@Override
public InternalAvg doReduce(List<InternalAggregation> aggregations, ReduceContext reduceContext) {
long count = 0;
double sum = 0;
for (InternalAggregation aggregation : aggregations) {
count += ((InternalAvg) aggregation).count;
sum += ((InternalAvg) aggregation).sum;
}
return new InternalAvg(getName(), sum, count, valueFormatter, pipelineAggregators(), getMetaData());
}
其邏輯相當簡單,count相加, sum相加。獲取最終的結果就是
public double getValue() {
return sum / count;
}
上面講述了ES分發會彙總的關鍵節點,那麼分發到各個節點的業務邏輯是怎樣的呢?
首先定位入口:
class SearchQueryTransportHandler extends TransportRequestHandler<ShardSearchTransportRequest> {
@Override
public void messageReceived(ShardSearchTransportRequest request, TransportChannel channel) throws Exception {
QuerySearchResultProvider result = searchService.executeQueryPhase(request);
channel.sendResponse(result);
}
}
``
然後定位到`QueryPhrase.execute()`, 在QueryPhrase這個階段,主要做的事情如下:
` aggregationPhase.preProcess(searchContext)`: 解析ES的語法,生成Collector.
`execute`: 在呼叫Lucene的介面查詢資料前,組合各個Collecotr, ` collector = MultiCollector.wrap(subCollectors);` 然後查詢Lucene索引。對於AvgAggregator, 其關鍵邏輯是:
@Override
public void collect(int doc, long bucket) throws IOException {
counts = bigArrays.grow(counts, bucket + 1);
sums = bigArrays.grow(sums, bucket + 1);
values.setDocument(doc);
final int valueCount = values.count();
counts.increment(bucket, valueCount);
double sum = 0;
for (int i = 0; i < valueCount; i++) {
sum += values.valueAt(i);
}
sums.increment(bucket, sum);
}
這個已經是第二次出現了, 它的功能就是收集每個命中查詢的doc相關資訊。 這裡獲取每個docId對應的value,是基於doc_value的正向索引。
以上就是整個Avg Aggregation的實現流程。 通過原始碼,可以確認, AvgAggregation是精確可信的。 還有幾個聚合函式,其思路跟AvgAggregation是一致的,就不細說了,他們分別是: Max, Min, Sum, ValueCount, Stats 。。。