ElasticSearch Aggregation Bucket 인스턴스 분석

앞에서 ElasticSearch Aggregations 분석에서 우리는 [Aggregation Bucket의 실현]을 언급했지만 문자로 원리를 간략하게 묘사했을 뿐이다.오늘 이 글은 간단한grouyby와 유사한 조작으로 여러분에게 Aggregator의 작업 원리를 한층 더 이해하게 할 것입니다

문의

오늘 저희가 가정한 조회는 다음과 같습니다.

{   
    "aggs":{ "user": { "terms": { "field": "user", "size": 10, "order": { "_count": "desc" } } } } }

그 의미는 이 ql문장과 유사합니다:selectcount(*)asuser_count group by user order by user_count desc. 사용자 필드에 따라 그룹by를 하고 이어서 내림차순으로 배열합니다.

호출 체인 관계

먼저

org.elasticsearch.search.aggregations.bucket.terms.TermsParser

위의 JSON 쿼리 문자열을 분석하여 구성합니다.

org.elasticsearch.search.aggregations.bucket.terms.TermsAggregatorFactory

이 Factory를 구축하려면 다음과 같은 정보가 필요합니다.

집합 이름

통계가 필요한 필드(ValueSourceConfig 객체)

정렬(Order 객체)

다른terms 집합 특유의 설정

나중에 만들어질 거예요.

org.elasticsearch.search.aggregations.bucket.terms.GlobalOrdinalsStringTermsAggregator

대상이것은 집합을 실현하는 핵심 대상이다. 물론 이런 유형은 현재의 예에만 적용된다.이 클래스에는 다음과 같은 두 가지 핵심 방법이 있습니다.

getLeafCollector

buildAggregation

getLeafCollector는 Collector를 가져오는 것입니다. 앞에서 언급한 Collector는 사실 모든 문서를 교체하는 교체기입니다. 이 단계에서 우리는 Collector를 가져오고 DocValues에 의존하여 계수를 합니다.
두 번째 방법,build Aggrgation 방법은 수집된 결과를 정렬하여 topN을 가져옵니다.

getLeafCollector

전체 방법의 코드를 나는 아래에 붙였다.원본 코드를 해석하는 과정에서 우리는 ES가 DocValues에 대한 봉인을 설명할 것입니다.

@Override
    public LeafBucketCollector getLeafCollector(LeafReaderContext ctx,
                                                final LeafBucketCollector sub) throws IOException {

       //valuesSource ：ValuesSource.Bytes.WithOrdinals.FieldData
        globalOrds = valuesSource.globalOrdinalsValues(ctx);

       //  if  ， 
        if (acceptedGlobalOrdinals == null && includeExclude != null) {
            acceptedGlobalOrdinals = includeExclude.acceptedGlobalOrdinals(globalOrds, valuesSource);
        }

        if (acceptedGlobalOrdinals != null) {
            globalOrds = new FilteredOrdinals(globalOrds, acceptedGlobalOrdinals);
        }
       //  collector
        return newCollector(globalOrds, sub);
    }

나는 앞의 원본 코드에 valuesSource의 유형을 특별히 주석했다.앞에서 언급한 바와 같이 대부분의 Aggregator는 FieldData/DocValues에 의존하여 이루어진 것이고 ValueSource는 그들이 ES에 표시한 것이다.그래서 그들을 이해하는 것이 필요하다.ValuesSource 전체 클래스 이름:

 org.elasticsearch.search.aggregations.support.ValuesSource

이 클래스는 ES가 DocValues를 관리하기 위해 봉인된 것입니다.이것은 추상적인 클래스로 내부에 아직도 많은 실현 클래스가 있다. Bytes, With Ordinals, Field Data, Numeric, Long Values 등이다.이러한 표현은 특정 유형의 DocValues 유형에 대한 ES 표현입니다.
위의 우리의 조회 예시를 보면 user 필드에 대응하는 것은

 org.elasticsearch.search.aggregations.support.ValuesSource.Bytes.WithOrdinals.FieldData

대상이 객체는 Lucene String 유형의 DocValues에 대한 ES의 표현입니다.ValueSource 클래스에서 FieldData가 다른 것을 발견할 수 있습니다.서로 다른 FieldData는 서로 다른 기류로부터 계승되어 서로 다른 유형의 데이터를 나타낼 수 있다.현재 이 FieldData에는 다음과 같은 객체가 있습니다.

protected final IndexOrdinalsFieldData indexFieldData;

이 대상은 user (우리의 예시 필드) 가 String 형식일 때, 실행 클래스는

org.elasticsearch.index.fielddata.plain.SortedSetDVOrdinalsIndexFieldData

이 객체의 대체적인 역할은 DocValue를 구성하는 ES의 Wraper입니다.
구체적인 코드는 다음과 같습니다.

@Overridepublic AtomicOrdinalsFieldData load(LeafReaderContext context) {    
return new SortedSetDVBytesAtomicFieldData(
   context.reader(),
   fieldNames.indexName());
}
// loadGlobal 
//org.elasticsearch.index.fielddata.ordinals.InternalGlobalOrdinalsIndexFieldData  

첫 번째 상황을 예로 들면 위의 코드 new는 새로운 org.elasticsearch.index.fielddata.AtomicOrdinalsFieldData 대상을 만들었고 이 대상의 실현 클래스는 SortedSetDVBytesAtomicFieldData이다.이 객체는 Lucene의 DocValues와 마지막 도킹을 완료합니다.

 @Override
    public RandomAccessOrds getOrdinalsValues() {
        try {
            return FieldData.maybeSlowRandomAccessOrds(DocValues.getSortedSet(reader, field));
        } catch (IOException e) {
            throw new IllegalStateException("cannot load docvalues", e);
        }
    }

Reader를 통해 얻은 마지막 열은 이 클래스의 get Ordinals Values 방법에서 이루어진 것입니다.
이 방법이 마지막으로 되돌아온 RandomAccessOrds는 바로 Lucene의 DocValues가 실현된 것이다.
이렇게 많이 분석한 결과 모든 논리는 getLeafCollector의 첫 줄 코드에 농축되었다.글로벌 Ords의 유형은 Random Access Ords이며 Lucene과 직접 대응합니다.

globalOrds = valuesSource.globalOrdinalsValues(cox);

getLeafCollector 마지막 newCollector 규칙은 다음과 같습니다.

 protected LeafBucketCollector newCollector(final RandomAccessOrds ords, final LeafBucketCollector sub) {
        grow(ords.getValueCount());
        final SortedDocValues singleValues = DocValues.unwrapSingleton(ords);
        if (singleValues != null) {
            return new LeafBucketCollectorBase(sub, ords) {
                @Override
                public void collect(int doc, long bucket) throws IOException {
                    assert bucket == 0;
                    final int ord = singleValues.getOrd(doc);
                    if (ord >= 0) {
                        collectExistingBucket(sub, doc, ord);
                    }
                }
            };
        }

우리는 Lucene에서 대부분의 파일이 업데이트할 수 없다는 것을 안다.한 단락이 생성되면 변하지 않는다. 새로운 데이터나 삭제 데이터는 모두 새로운 단락을 생성해야 한다.DocValues의 스토리지 파일도 비슷합니다.그래서 DocValues.unwrap Singleton은 사실 이 판정을 한 것이다. 여러 개의 파일이 있는 것 아니냐.아니든 아니든 익명의 Collector를 직접 만들었습니다.
파일이 잘 이해되면 색인에 있는user 필드의 모든 값을 포함하고 좌표를 얻는 것도 자연스럽습니다.

//singleValues RandomAccessOrds。
final int ord = singleValues.getOrd(doc);

문서 번호에 따라 값을 가져오는 위치에 따라 ord >=0이면 값이 있고, 그렇지 않으면 값이 없습니다.
파일이 여러 개 있으면 다음과 같은 Collecor가 반환됩니다.

else {
            return new LeafBucketCollectorBase(sub, ords) {
                @Override
                public void collect(int doc, long bucket) throws IOException {
                    assert bucket == 0;
                    ords.setDocument(doc);
                    final int numOrds = ords.cardinality();
                    for (int i = 0; i < numOrds; i++) {
                        final long globalOrd = ords.ordAt(i);
                        collectExistingBucket(sub, doc, globalOrd);
                    }
                }
            };

위의 코드는 여러 개의 파일을 최종적으로 합쳐서 하나의 파일의 번호를 유지할 수 있도록 보장할 수 있다.무슨 뜻이죠?예를 들어 A 파일에 문서가 하나 있고 B 파일에 문서가 하나 있다면 최종적으로 얻은 global Ord는 0, 1이지 모두 0이 아니다.이때 ords 구현 클래스는 SingletonSortedSetDocValues가 아니라

org.elasticsearch.index.fielddata.ordinals.GlobalOrdinalMapping

상대가 되다.
계수의 방식은 둘 다 대체적으로 유사하다.

docCounts.increment(bucketOrd, 1);

이곳의 버킷 오드는 사실 앞의 ord/global Ord입니다.그래서 전체 계산은 docCounts를 채우는 거예요.

buildAggregation

buildAggregation 코드는 붙이지 않겠습니다. 대체적인 논리는:

user 필드가 없으면 바로 되돌아옵니다.

반환값의 크기를 가져옵니다. TopN의 N

우선 순위 대기열을 구축하여 정렬 최고terms

를 가져옵니다.

StringTerms 구축 반환

이 논리도 Terms Aggrator의 전역 정렬이 정확하지 않다는 문제를 반영했다.각 Shard는 TopN 개만 가져오기 때문에 마지막에 Merge(Reduce) 다음에 반드시 전체 TopN이 아니기 때문입니다.너는 사이즈를 통해 정확도를 조절할 수 있다.
사실 한 측면에서 검증한 바와 같이 집합 결과가 비교적 큰 상황에서 ES는 한계가 있다.가장 좋은 방안은 ES가 모든termcount를 계산한 다음iterator를 통해 스파크에 접속하는 것이다. 스파크는 이 데이터를 불러와 마지막 전역 정렬을 하는 것이 좋다.

총결산

우리는 간단한 예로 Global Ordinals String Terms Aggregator를 분석하고 ES가 Lucene DocValues에 대한 봉인 방식을 언급했다.