[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進行清除。

1. /**
2. * 清除滾動ID
3. * @param client
4. * @param scrollIdList
5. * @return
6. */
7. public static boolean clearScroll(Client client, List<String> scrollIdList){
8. ClearScrollRequestBuilder clearScrollRequestBuilder = client.prepareClearScroll();
9. clearScrollRequestBuilder.setScrollIds(scrollIdList);
10. ClearScrollResponse response = clearScrollRequestBuilder.get();
11. return response.isSucceeded();
12. }
13. /**
14. * 清除滾動ID
15. * @param client
16. * @param scrollId
17. * @return
18. */
19. public static boolean clearScroll(Client client, String scrollId){
20. ClearScrollRequestBuilder clearScrollRequestBuilder = client.prepareClearScroll();
21. clearScrollRequestBuilder.addScrollId(scrollId);
22. ClearScrollResponse response = clearScrollRequestBuilder.get();
23. return response.isSucceeded();
24. }

4. 參考：

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

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

---------------------------------------

原文地址：

　　https://blog.csdn.net/u014589856/article/details/78775233