[ElasticSearch]Java API 之 滚动搜索(Scroll API)

一般搜索请求都是返回一"页"数据，无论数据量多大都一起返回给用户，Scroll API可以允许我们检索大量数据（甚至全部数据）。Scroll API允许我们做一个初始阶段搜索并且持续批量从Elasticsearch里拉取结果直到没有结果剩下。这有点像传统数据库里的cursors（游标）。

Scroll API的创建并不是为了实时的用户响应，而是为了处理大量的数据（Scrolling is not intended for real time user requests, but rather for processing large amounts of data）。从 scroll 请求返回的结果只是反映了 search 发生那一时刻的索引状态，就像一个快照(The results that are returned from a scroll request reflect the state of the index at the time that the initial search request was made, like a snapshot in time)。后续的对文档的改动（索引、更新或者删除）都只会影响后面的搜索请求。

1. 普通请求

假设我们想一次返回大量数据，下面代码中一次请求58000条数据：

1.    /**
2.     *  普通搜索
3.     * @param client
4.     */
5.    public static void search(Client client) {
7.        String index = "simple-index";
8.        String type = "simple-type";
10.        // 搜索条件
11.        SearchRequestBuilder searchRequestBuilder = client.prepareSearch();
12.        searchRequestBuilder.setIndices(index);
13.        searchRequestBuilder.setTypes(type);
14.        searchRequestBuilder.setSize(58000);
16.        // 执行
17.        SearchResponse searchResponse = searchRequestBuilder.get();
18.        // 搜索结果
19.        SearchHit[] searchHits = searchResponse.getHits().getHits();
20.        for (SearchHit searchHit : searchHits) {
21.            String source = searchHit.getSource().toString();
22.            logger.info("--------- searchByScroll source {}", source);
23.        } // for
25.    }

运行结果：

1. Caused by: QueryPhaseExecutionException[Result window is too large, from + size must be less than or equal to: [10000] but was [58000]. See the scroll api for a more efficient way to request large data sets. This limit can be set by changing the [index.max_result_window] index level parameter.]
2. at org.elasticsearch.search.internal.DefaultSearchContext.preProcess(DefaultSearchContext.java:212)
3. at org.elasticsearch.search.query.QueryPhase.preProcess(QueryPhase.java:103)
4. at org.elasticsearch.search.SearchService.createContext(SearchService.java:676)
5. at org.elasticsearch.search.SearchService.createAndPutContext(SearchService.java:620)
6. at org.elasticsearch.search.SearchService.executeQueryPhase(SearchService.java:371)
7. at org.elasticsearch.search.action.SearchServiceTransportAction$SearchQueryTransportHandler.messageReceived(SearchServiceTransportAction.java:368)
8. at org.elasticsearch.search.action.SearchServiceTransportAction$SearchQueryTransportHandler.messageReceived(SearchServiceTransportAction.java:365)
9. at org.elasticsearch.transport.TransportRequestHandler.messageReceived(TransportRequestHandler.java:33)
10. at org.elasticsearch.transport.RequestHandlerRegistry.processMessageReceived(RequestHandlerRegistry.java:75)
11. at org.elasticsearch.transport.TransportService$4.doRun(TransportService.java:376)
12. at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:37)
13. ... 3 more

从上面我们可以知道，搜索请求一次请求最大量为[10000]。我们的请求量已经超标，因此报错，异常信息提示我们请求大数据量的情况下使用Scroll API。

2. 使用Scroll API 请求

为了使用 scroll，初始搜索请求应该在查询中指定 scroll 参数，告诉 Elasticsearch 需要保持搜索的上下文环境多长时间（滚动时间）。

1. searchRequestBuilder.setScroll(new TimeValue(60000));

下面代码中指定了查询条件以及滚动属性，如滚动的有效时长（使用setScroll()方法）。我们通过SearchResponse对象的getScrollId()方法获取滚动ID。滚动ID会在下一次请求中使用。

1.    /**
2.     * 使用scroll进行搜索
3.     * @param client
4.     */
5.    public static String searchByScroll(Client client) {
7.        String index = "simple-index";
8.        String type = "simple-type";
10.        // 搜索条件
11.        SearchRequestBuilder searchRequestBuilder = client.prepareSearch();
12.        searchRequestBuilder.setIndices(index);
13.        searchRequestBuilder.setTypes(type);
14.        searchRequestBuilder.setScroll(new TimeValue(30000));
16.        // 执行
17.        SearchResponse searchResponse = searchRequestBuilder.get();
19.        String scrollId = searchResponse.getScrollId();
20.        logger.info("--------- searchByScroll scrollID {}", scrollId);
22.        SearchHit[] searchHits = searchResponse.getHits().getHits();
23.        for (SearchHit searchHit : searchHits) {
24.            String source = searchHit.getSource().toString();
25.            logger.info("--------- searchByScroll source {}", source);
26.        } // for
27.        return scrollId;
28.        
29.    }

使用上面的请求返回的结果中的滚动ID，这个 ID 可以传递给 scroll API 来检索下一个批次的结果。这一次请求中不用添加索引和类型，这些都指定在了原始的 search 请求中。

每次返回下一个批次结果 直到没有结果返回时停止 即hits数组空时(Each call to the scroll API returns the next batch of results until there are no more results left to return, ie the hits array is empty)。

1.    /**
2.     *  通过滚动ID获取文档
3.     * @param client
4.     * @param scrollId
5.     */
6.    public static void searchByScrollId(Client client, String scrollId){
9.        TimeValue timeValue = new TimeValue(30000);
10.        SearchScrollRequestBuilder searchScrollRequestBuilder;
11.        SearchResponse response;
12.        // 结果
13.        while (true) {
15.            logger.info("--------- searchByScroll scrollID {}", scrollId);
17.            searchScrollRequestBuilder = client.prepareSearchScroll(scrollId);
18.            // 重新设定滚动时间
19.            searchScrollRequestBuilder.setScroll(timeValue);
20.            // 请求
21.            response = searchScrollRequestBuilder.get();
22.            // 每次返回下一个批次结果 直到没有结果返回时停止 即hits数组空时
23.            if (response.getHits().getHits().length == 0) {
24.                break;
25.            } // if
26.            // 这一批次结果
27.            SearchHit[] searchHits = response.getHits().getHits();
28.            for (SearchHit searchHit : searchHits) {
29.                String source = searchHit.getSource().toString();
30.                logger.info("--------- searchByScroll source {}", source);
31.            } // for
32.            // 只有最近的滚动ID才能被使用
33.            scrollId = response.getScrollId();
35.        } // while
37.    }

备注：

初始搜索请求和每个后续滚动请求返回一个新的 滚动ID——只有最近的滚动ID才能被使用。（The initial search request and each subsequent scroll request returns a new_scroll_id — only the most recent _scroll_id should be used）  

我每次后续滚动请求返回的滚动ID都是相同的，所以对上面的备注，不是很懂，有明白的可以告知，谢谢。

如果超过滚动时间，继续使用该滚动ID搜索数据，则会报错：

1. Caused by: SearchContextMissingException[No search context found for id [2861]]
2. at org.elasticsearch.search.SearchService.findContext(SearchService.java:613)
3. at org.elasticsearch.search.SearchService.executeQueryPhase(SearchService.java:403)
4. at org.elasticsearch.search.action.SearchServiceTransportAction$SearchQueryScrollTransportHandler.messageReceived(SearchServiceTransportAction.java:384)
5. at org.elasticsearch.search.action.SearchServiceTransportAction$SearchQueryScrollTransportHandler.messageReceived(SearchServiceTransportAction.java:381)
6. at org.elasticsearch.transport.TransportRequestHandler.messageReceived(TransportRequestHandler.java:33)
7. at org.elasticsearch.transport.RequestHandlerRegistry.processMessageReceived(RequestHandlerRegistry.java:75)
8. at org.elasticsearch.transport.TransportService$4.doRun(TransportService.java:376)
9. at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:37)
10. at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
11. at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
12. at java.lang.Thread.run(Thread.java:745)

3. 清除滚动ID

虽然当滚动有效时间已过，搜索上下文(Search Context)会自动被清除，但是一值保持滚动代价也是很大的，所以当我们不在使用滚动时要尽快使用Clear-Scroll API进行清除。

4. 参考：

https://www.elastic.co/guide/en/elasticsearch/reference/2.4/search-request-scroll.html

http://www.jianshu.com/p/14aa8b09c789

5. 说明

本代码基于ElasticSearch 2.4.1