专栏首页YG小书屋ES5.6 search流程与scroll

ES5.6 search流程与scroll

ES search大致流程

请求转化,由RestSearchAction转为TransportSearchAction,执行其doExecute()方法

1、如果查询请求的索引含有正则表达式和别名,找出具体的索引

indices = indexNameExpressionResolver.concreteIndices(clusterState, searchRequest.indicesOptions(),
            timeProvider.getAbsoluteStartMillis(), localIndices.indices());

2、找出routing,并根据routing找出shard

Map<String, Set<String>> routingMap = indexNameExpressionResolver.resolveSearchRouting(clusterState, searchRequest.routing(),
      searchRequest.indices());
GroupShardsIterator<ShardIterator> localShardsIterator = clusterService.operationRouting().searchShards(clusterState,
        concreteIndices, routingMap, searchRequest.preference());

3、解析请求中的权重boost

Map<String, Float> concreteIndexBoosts = resolveIndexBoosts(searchRequest, clusterState);

4、根据请求的类型选择不同的query类

switch(searchRequest.searchType()) {
    case DFS_QUERY_THEN_FETCH:
        searchAsyncAction = new SearchDfsQueryThenFetchAsyncAction(logger, searchTransportService, connectionLookup,
            aliasFilter, concreteIndexBoosts, searchPhaseController, executor, searchRequest, listener, shardIterators,
            timeProvider, clusterStateVersion, task);
        break;
    case QUERY_AND_FETCH:
    case QUERY_THEN_FETCH:
        searchAsyncAction = new SearchQueryThenFetchAsyncAction(logger, searchTransportService, connectionLookup,
            aliasFilter, concreteIndexBoosts, searchPhaseController, executor, searchRequest, listener, shardIterators,
            timeProvider, clusterStateVersion, task);
        break;
    default:
        throw new IllegalStateException("Unknown search type: [" + searchRequest.searchType() + "]");
}

插播一条DFS_QUERY_THEN_FETCH与QUERY_THEN_FETCH的区别。

  • QUERY_THEN_FETCH 请求执行有两个阶段。第一个阶段就是查询所有相关的shards。所有的shards执行请求并根据form和size返回一个排好序的结果。协调节点获取到所有的信息后merge并resort,然后根据form和size取出结果。第二个阶段则是根据第一阶段的结果在相应的shard上取出数据。
  • DFS_QUERY_THEN_FETCH 与QUERY_THEN_FETCH几乎一样。只有在第一阶段为了更精确的打分,计算的是分布式的term frequencies。

5、SearchQueryThenFetchAsyncAction中则是执行标准的查询流程:

  • query
  • fetch
  • merge

SearchQueryThenFetchAsyncAction标准查询流程

1、Query

首先遍历shard,执行shard查询请求。

for (final SearchShardIterator shardIt : shardsIts) {
    shardIndex++;
    final ShardRouting shard = shardIt.nextOrNull();
    if (shard != null) {
        performPhaseOnShard(shardIndex, shardIt, shard);
    } else {
        // really, no shards active in this group
        onShardFailure(shardIndex, null, null, shardIt, new NoShardAvailableActionException(shardIt.shardId()));
    }
}

每个shard通过SearchTransportService发送query请求:

getSearchTransport().sendExecuteQuery(getConnection(shardIt.getClusterAlias(), shard.currentNodeId()),
        buildShardSearchRequest(shardIt), getTask(), listener);

本地则是执行TransportService的sendLocalRequest方法.最终调用的是

handler.messageReceived(request, channel);

此处的channel是DirectResponseChannel,而此处的handler则是TransportSearchAction 初始化的时候写入SearchTransportService中的。

transportService.registerRequestHandler(QUERY_ACTION_NAME, ShardSearchTransportRequest::new, ThreadPool.Names.SEARCH,
  new TaskAwareTransportRequestHandler<ShardSearchTransportRequest>() {
    @Override
    public void messageReceived(ShardSearchTransportRequest request, TransportChannel channel, Task task) throws Exception {
        SearchPhaseResult result = searchService.executeQueryPhase(request, (SearchTask)task);
        channel.sendResponse(result);
    }
});

因此,最终执行的是searchService的executeQueryPhase方法。

真正的query本地查询:

1)生成一个searchContext,然后将request写入context:

final DefaultSearchContext searchContext = new DefaultSearchContext(idGenerator.incrementAndGet(), request, shardTarget,
        engineSearcher, indexService, indexShard, bigArrays, threadPool.estimatedTimeInMillisCounter(), timeout, fetchPhase);

parseSource(context, request.source());

2)queryPhase预处理。比如说如果请求没有query,默认为match_all,如果没有boost,采用默认的值等等

queryPhase.preProcess(context);

3)执行查询

loadOrExecuteQueryPhase(request, context);

loadOrExecuteQueryPhase中则是调用queryPhase执行查询:

queryPhase.execute(context);

在queryPhase中执行真正的query查询: 1)重写query,比如说 讲一个prefixQuery转化为多个包含termQuery的BoolQuery assert query == searcher.rewrite(query); // already rewritten

2)将各个子查询求用Collector包裹,Collector传给lucene,进行真正的lucene查询

if (terminateAfterSet) {
    final Collector child = collector;
    // throws Lucene.EarlyTerminationException when given count is reached
    collector = Lucene.wrapCountBasedEarlyTerminatingCollector(collector, searchContext.terminateAfter());
    if (doProfile) {
        collector = new InternalProfileCollector(collector, CollectorResult.REASON_SEARCH_TERMINATE_AFTER_COUNT,
                Collections.singletonList((InternalProfileCollector) child));
    }
}

if (searchContext.parsedPostFilter() != null) {
    final Collector child = collector;
    // this will only get applied to the actual search collector and not
    // to any scoped collectors, also, it will only be applied to the main collector
    // since that is where the filter should only work
    final Weight filterWeight = searcher.createNormalizedWeight(searchContext.parsedPostFilter().query(), false);
    collector = new FilteredCollector(collector, filterWeight);
    if (doProfile) {
        collector = new InternalProfileCollector(collector, CollectorResult.REASON_SEARCH_POST_FILTER,
                Collections.singletonList((InternalProfileCollector) child));
    }
}
// plug in additional collectors, like aggregations
final List<Collector> subCollectors = new ArrayList<>();
subCollectors.add(collector);
subCollectors.addAll(searchContext.queryCollectors().values());
collector = MultiCollector.wrap(subCollectors);

//最后包裹一个Cancellable
if (collector != null) {
    final Collector child = collector;
    collector = new CancellableCollector(searchContext.getTask()::isCancelled, searchContext.lowLevelCancellation(), collector);
    if (doProfile) {
        collector = new InternalProfileCollector(collector, CollectorResult.REASON_SEARCH_CANCELLED,
            Collections.singletonList((InternalProfileCollector) child));
    }
}

3)执行lucene的查询

searcher.search(query, collector);

4)将查询结果以TopDocs的方式返回

queryResult.topDocs(topDocsCallable.call(), sortValueFormats);

topDocsCallable是根据不同的查询封装的。举个例子:

if (query.getClass() == MatchAllDocsQuery.class) {
    collector = null;
    topDocsCallable = new Callable<TopDocs>() {
        @Override
        public TopDocs call() throws Exception {
            int count = searcher.getIndexReader().numDocs();
            return new TopDocs(count, Lucene.EMPTY_SCORE_DOCS, 0);
        }
    };
} else if (query.getClass() == TermQuery.class && searcher.getIndexReader().hasDeletions() == false) {
    final Term term = ((TermQuery) query).getTerm();
    collector = null;
    topDocsCallable = new Callable<TopDocs>() {
        @Override
        public TopDocs call() throws Exception {
            int count = 0;
            for (LeafReaderContext context : searcher.getIndexReader().leaves()) {
                count += context.reader().docFreq(term);
            }
            return new TopDocs(count, Lucene.EMPTY_SCORE_DOCS, 0);
        }
    };
}

2、fetch阶段(+merge阶段):

执行的是FetchSearchPhase的dorun方法。 1)首先reduce

final SearchPhaseController.ReducedQueryPhase reducedQueryPhase = resultConsumer.reduce();

resultConsumer 是在构造SearchQueryThenFetchAsyncAction的时候创建的,执行的是SearchPhaseController的reducedQueryPhase方法。

searchPhaseController.newSearchPhaseResults(request, shardsIts.size())

2)查找reduce出的文档id,按照shardId存放

final IntArrayList[] docIdsToLoad = searchPhaseController.fillDocIdsToLoad(numShards, reducedQueryPhase.scoreDocs);

如果是scroll查询,将lastEmittedDocPerShard存入ShardFetchSearchRequest:

final ScoreDoc[] lastEmittedDocPerShard = isScrollSearch ?
               searchPhaseController.getLastEmittedDocPerShard(reducedQueryPhase, numShards)
               : null;
ShardFetchSearchRequest fetchSearchRequest = createFetchRequest(queryResult.queryResult().getRequestId(), i, entry,
                            lastEmittedDocPerShard, searchShardTarget.getOriginalIndices());

3)针对于每个shard,创建fetch请求,并执行

ShardFetchSearchRequest fetchSearchRequest = createFetchRequest(queryResult.queryResult().getRequestId(), i, entry,
    lastEmittedDocPerShard, searchShardTarget.getOriginalIndices());
executeFetch(i, searchShardTarget, counter, fetchSearchRequest, queryResult.queryResult(),
    connection);

本地节点收到请求后执行的是SearchService的executeFetchPhase方法。这主要涉及些lucene的东西,不在详细解析。 4)当从所有的shard完获取完结果之后执行:

final Runnable finishPhase = ()
-> moveToNextPhase(searchPhaseController, scrollId, reducedQueryPhase, queryAndFetchOptimization ?
queryResults : fetchResults);

moveToNextPhase中开始执行merge:

final InternalSearchResponse internalResponse = searchPhaseController.merge(context.getRequest().scroll() != null,
reducedQueryPhase, fetchResultsArr.asList(), fetchResultsArr::get);

merge结束后,通过nextPhaseFactory将结果发送出去:

    (response, scrollId) -> new ExpandSearchPhase(context, response,
                (finalResponse) -> sendResponsePhase(finalResponse, scrollId, context))

  private static SearchPhase sendResponsePhase(InternalSearchResponse response, String scrollId, SearchPhaseContext context) {
    return new SearchPhase("response") {
        @Override
        public void run() throws IOException {
            context.onResponse(context.buildSearchResponse(response, scrollId));
        }
    };
}

context则是SearchQueryThenFetchAsyncAction的父类AbstractSearchAsyncAction,调用buildSearchResponse方法构造response,并回调listener。

public final void onResponse(SearchResponse response) {
    listener.onResponse(response);
}

scroll查询

scroll查询原理

在第一次查询时,记录上一次查询的位置,在接下来的查询中获取到上次查询的位置,接着查询。

比如说将查询order by time offset 0 limit 100,改写成order by time where time>0 limit 100,记录最后一个$time_max,接下来的查询order by time offset 100 limit 100,改写成order by time where time>$time_max limit 100。如此往复,可以看出scroll是一个常量查询延迟和开销。 这个从源码中也可以看出:

if (returnsDocsInOrder(query, searchContext.sort())) {
    if (scrollContext.totalHits == -1) {
        // first round
        assert scrollContext.lastEmittedDoc == null;
        // there is not much that we can optimize here since we want to collect all
        // documents in order to get the total number of hits
    } else {
        // now this gets interesting: since we sort in index-order, we can directly
        // skip to the desired doc and stop collecting after ${size} matches
        if (scrollContext.lastEmittedDoc != null) {
        if (scrollContext.lastEmittedDoc != null) {
            BooleanQuery bq = new BooleanQuery.Builder()
                .add(query, BooleanClause.Occur.MUST)
                .add(new MinDocQuery(after.doc + 1), BooleanClause.Occur.FILTER)
                .build();
            query = bq;
        }
        searchContext.terminateAfter(numDocs);
    }
}

当scrollContext.lastEmittedDoc不为空也就是查询scroll的下一页时,将查询变为一个带偏移量的bool查询。 在本地查出数据后,更新scrollContext的值:

if (scrollContext != null) {
    if (scrollContext.totalHits == -1) {
        // first round
        scrollContext.totalHits = topDocs.totalHits;
        scrollContext.maxScore = topDocs.getMaxScore();
    } else {
        // subsequent round: the total number of hits and
        // the maximum score were computed on the first round
        topDocs.totalHits = scrollContext.totalHits;
        topDocs.setMaxScore(scrollContext.maxScore);
    }
    if (searchContext.request().numberOfShards() == 1) {
        // if we fetch the document in the same roundtrip, we already know the last emitted doc
        if (topDocs.scoreDocs.length > 0) {
            // set the last emitted doc
            scrollContext.lastEmittedDoc = topDocs.scoreDocs[topDocs.scoreDocs.length - 1];
        }
    }
}

我们看到上述只更新了scrollContext的totalHits和maxScore,而lastEmittedDoc则是在fetch阶段中更新的: 先将lastEmittedDocPerShard存入ShardFetchSearchRequest:

final ScoreDoc[] lastEmittedDocPerShard = isScrollSearch ?
               searchPhaseController.getLastEmittedDocPerShard(reducedQueryPhase, numShards)
               : null;
ShardFetchSearchRequest fetchSearchRequest = createFetchRequest(queryResult.queryResult().getRequestId(), i, entry,
                            lastEmittedDocPerShard, searchShardTarget.getOriginalIndices());

然后每个节点在接收到fetch信息后,执行SearchService的executeFetchPhase方法:

 if (request.lastEmittedDoc() != null) {
                context.scrollContext().lastEmittedDoc = request.lastEmittedDoc();
 }

当使用scroll请求时,scrollID是不变的,从SearchScrollAsyncAction代码中可以看出:

protected final void sendResponse(SearchPhaseController.ReducedQueryPhase queryPhase,
                                  final AtomicArray<? extends SearchPhaseResult> fetchResults) {
    try {
        final InternalSearchResponse internalResponse = searchPhaseController.merge(true, queryPhase, fetchResults.asList(),
            fetchResults::get);
        // the scroll ID never changes we always return the same ID. This ID contains all the shards and their context ids
        // such that we can talk to them abgain in the next roundtrip.
        String scrollId = null;
        if (request.scroll() != null) {
            scrollId = request.scrollId();
        }
        listener.onResponse(new SearchResponse(internalResponse, scrollId, this.scrollId.getContext().length, successfulOps.get(),
            buildTookInMillis(), buildShardFailures()));
    } catch (Exception e) {
        listener.onFailure(new ReduceSearchPhaseException("fetch", "inner finish failed", e, buildShardFailures()));
    }
}

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • python 进程池异步调用与进程间通信

    YG
  • ES存在unassinged shard的调试方式

    YG
  • 【转】架构漫谈(二):认识概念是理解架构的基础

    架构漫谈是由资深架构师王概凯 Kevin 执笔的系列专栏,专栏将会以 Kevin 的架构经验为基础,逐步讨论什么是架构、怎样做好架构、软件架构如何落地、如何写好...

    YG
  • Netty NioEventLoop源码解读

      NioEventLoop中维护了一个线程,线程启动时会调用NioEventLoop的run方法,执行I/O任务和非I/O任务:I/O任务:即selectio...

    良辰美景TT
  • PAT Advanced 1065

    Given three integers A, B and C in [-263, 263], you are supposed to tell whethe...

    chain
  • 聊聊springboot2的micrometer

    springboot2在spring-boot-actuator中引入了micrometer,对1.x的metrics进行了重构,另外支持对接的监控系统也更加丰...

    codecraft
  • 【JavaWeb】64:多表查询详解

    万万没有想到,学个数据库竟然还能接触到笛卡尔积?后面不会学着学着还会出现牛顿吧……

    刘小爱
  • 在阿里“解放”鉴黄师是一种怎样的体验

    2009 年,主要从事数字信号和图像处理研究、博士刚刚毕业的薛晖拿到了阿里巴巴的 offer,那一年淘宝的全年交易额已经达到 2083 亿元,薛晖激动地搓搓小手...

    昱良
  • sublime 3及常用插件

    https://www.baidu.com/link?url=c9YyfvWOfn0EtUrhKlZQ26ANUOD_CSqjgqqsb3lq6LQ05oy2M...

    py3study
  • 重新设置Eclipse的workspace路径

    1. 启动Eclipse/MyEclipse后, 打开"Window -> Preferences -> General -> Startup and Shut...

    ydymz

扫码关注云+社区

领取腾讯云代金券