[ElasticSearch]Java API 之滚动搜索(Scroll API)-低调大师

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

2016-10-12 549

一般搜索请求都是返回一"页"数据，无论数据量多大都一起返回给用户，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条数据：

/**
* 普通搜索
* @param client
*/
public static void search(Client client) {
String index = "simple-index";
String type = "simple-type";
// 搜索条件
SearchRequestBuilder searchRequestBuilder = client.prepareSearch();
searchRequestBuilder.setIndices(index);
searchRequestBuilder.setTypes(type);
searchRequestBuilder.setSize(58000);
// 执行
SearchResponse searchResponse = searchRequestBuilder.get();
// 搜索结果
SearchHit[] searchHits = searchResponse.getHits().getHits();
for (SearchHit searchHit : searchHits) {
String source = searchHit.getSource().toString();
logger.info("--------- searchByScroll source {}", source);
} // for
}

运行结果：

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.]
at org.elasticsearch.search.internal.DefaultSearchContext.preProcess(DefaultSearchContext.java:212)
at org.elasticsearch.search.query.QueryPhase.preProcess(QueryPhase.java:103)
at org.elasticsearch.search.SearchService.createContext(SearchService.java:676)
at org.elasticsearch.search.SearchService.createAndPutContext(SearchService.java:620)
at org.elasticsearch.search.SearchService.executeQueryPhase(SearchService.java:371)
at org.elasticsearch.search.action.SearchServiceTransportAction$SearchQueryTransportHandler.messageReceived(SearchServiceTransportAction.java:368)
at org.elasticsearch.search.action.SearchServiceTransportAction$SearchQueryTransportHandler.messageReceived(SearchServiceTransportAction.java:365)
at org.elasticsearch.transport.TransportRequestHandler.messageReceived(TransportRequestHandler.java:33)
at org.elasticsearch.transport.RequestHandlerRegistry.processMessageReceived(RequestHandlerRegistry.java:75)
at org.elasticsearch.transport.TransportService$4.doRun(TransportService.java:376)
at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:37)
... 3 more

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

2. 使用Scroll API 请求

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

searchRequestBuilder.setScroll(new TimeValue(60000));

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

/**
* 使用scroll进行搜索
* @param client
*/
public static String searchByScroll(Client client) {
String index = "simple-index";
String type = "simple-type";
// 搜索条件
SearchRequestBuilder searchRequestBuilder = client.prepareSearch();
searchRequestBuilder.setIndices(index);
searchRequestBuilder.setTypes(type);
searchRequestBuilder.setScroll(new TimeValue(30000));
// 执行
SearchResponse searchResponse = searchRequestBuilder.get();
String scrollId = searchResponse.getScrollId();
logger.info("--------- searchByScroll scrollID {}", scrollId);
SearchHit[] searchHits = searchResponse.getHits().getHits();
for (SearchHit searchHit : searchHits) {
String source = searchHit.getSource().toString();
logger.info("--------- searchByScroll source {}", source);
} // for
return scrollId;
}

使用上面的请求返回的结果中的滚动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)。

/**
* 通过滚动ID获取文档
* @param client
* @param scrollId
*/
public static void searchByScrollId(Client client, String scrollId){
TimeValue timeValue = new TimeValue(30000);
SearchScrollRequestBuilder searchScrollRequestBuilder;
SearchResponse response;
// 结果
while (true) {
logger.info("--------- searchByScroll scrollID {}", scrollId);
searchScrollRequestBuilder = client.prepareSearchScroll(scrollId);
// 重新设定滚动时间
searchScrollRequestBuilder.setScroll(timeValue);
// 请求
response = searchScrollRequestBuilder.get();
// 每次返回下一个批次结果直到没有结果返回时停止即hits数组空时
if (response.getHits().getHits().length == 0) {
break;
} // if
// 这一批次结果
SearchHit[] searchHits = response.getHits().getHits();
for (SearchHit searchHit : searchHits) {
String source = searchHit.getSource().toString();
logger.info("--------- searchByScroll source {}", source);
} // for
// 只有最近的滚动ID才能被使用
scrollId = response.getScrollId();
} // while
}

备注：

初始搜索请求和每个后续滚动请求返回一个新的滚动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搜索数据，则会报错：

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

3. 清除滚动ID

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

/**
* 清除滚动ID
* @param client
* @param scrollIdList
* @return
*/
public static boolean clearScroll(Client client, List<String> scrollIdList){
ClearScrollRequestBuilder clearScrollRequestBuilder = client.prepareClearScroll();
clearScrollRequestBuilder.setScrollIds(scrollIdList);
ClearScrollResponse response = clearScrollRequestBuilder.get();
return response.isSucceeded();
}
/**
* 清除滚动ID
* @param client
* @param scrollId
* @return
*/
public static boolean clearScroll(Client client, String scrollId){
ClearScrollRequestBuilder clearScrollRequestBuilder = client.prepareClearScroll();
clearScrollRequestBuilder.addScrollId(scrollId);
ClearScrollResponse response = clearScrollRequestBuilder.get();
return response.isSucceeded();
}

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

微信关注我们

原文链接：https://yq.aliyun.com/articles/632233

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

开源大数据周刊-第25期

阿里云E-Mapreduce动态 E-Mapreduce团队 1.6.0版本交互式查询（支持hive、spark）资讯阿里、Databricks、Intel的十位专家和你细聊Hadoop技术实践及生态2016年10月16日云栖小镇 D5厅不见不散大数据领域开源技术除了Hadoop你还知道哪些众所周知，大数据正在以惊人的速度增长，几乎触及各行各业，许多组织都被迫寻找新的创造性方法来管理和控制如此庞大的数据，当然这么做的目的不只是管理和控制数据，而是要分析和挖掘其中的价值，来促进业务的发展我们该如何看待空间大数据的未来？4D数据是测绘地理信息人的全部吗?答案不应该是肯定的。过去，我们对传统地信企业和部门，尤其是诸如测绘院等传统机构的印象，多停留在地理国情普查等庞大的职能上，我们认为他们应该把握最全面和精确的数据，那么除此之外呢大数据

2016-10-13

650

Mahout 是一个很强大的数据挖掘工具，是一个分布式机器学习算法的集合，包括被称为Taste的分布式协同过滤的实现、分类、聚类等。Mahout最大的优点就是基于hadoop实现，把很多以前运行于单机上的算法，转化为了MapReduce模式，这样大大提升了算法可处理的数据量和处理性能。 http://www.ha97.com/5803.html 1.下载解压[root@sht-sgmhadoopnn-01 hadoop]# wget http://apache.fayea.com/mahout/0.12.2/apache-mahout-distribution-0.12.2.tar.gz[root@sht-sgmhadoopnn-01 hadoop]# tar -xzvf apache-mahout-distribution-0.12.2.tar.gz [root@sht-sgmhadoopnn-01 hadoop]# ln -s /hadoop/apache-mahout-distribution-0.12.2 mahout 2.配置环境变量[root@sht-sgmhadoopnn...

2016-10-13

811

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。