Elasticsearch使用Scroll-Scan实现数据遍历

Myth丶恋晨 2022-06-12 01:56 313阅读 0赞

Elasticsearch 是一个实时的分布式搜索与分析引擎，被广泛用来做全文搜索、结构化搜索、结果分析。在实际应用中有时需要遍历某个索引的全部数据，这时使用分页检索的形式效率会比较差。

分页检索即from-size形式，from指的是从哪里开始拿数据，size是结果集中返回的文档个数。from-size的工作原理是：如size=10&from=100，那么Elasticsearch会从每个分片里取出110条数据，然后汇集到一起再排序，取出101~110序号的文档。由此可见，from-size的效率必然不会很高，特别是分页越深，需要排序的数据越多，其效率就越低。

这时更为有效的方法是使用Scroll-Scan。Scroll是先做一次初始化搜索把所有符合搜索条件的结果缓存起来生成一个快照，然后持续地、批量地从快照里拉取数据直到没有数据剩下。而这时对索引数据的插入、删除、更新都不会影响遍历结果，因此scroll 并不适合用来做实时搜索。Scan是搜索类型，告诉Elasticsearch不用对结果集进行排序，只要分片里还有结果可以返回，就返回一批结果。scroll- scan使用中不能跳页获取结果，必须一页接着一页获取。

为了使用scroll-scan，需要执行一个初始化搜索请求，将search\_type设置成scan，并且传递一个scroll参数来告诉 Elasticsearch缓存应该持续多长时间，在缓存持续时间内初始化搜索请求后对索引的修改不会反应到快照中。每次搜索请求后都会返回一个scrollId，是一个 64 位的字符串编码，后续会使用此scrollId来获取数据。scroll时间指的是本次数据处理所需要的时间，如果超过此时间，继续使用该scrollId搜索数据则会报错。在使用scroll-scan时可以指定返回结果集大小，在 scan 的时候，size 作用在每个分片上，所以将会在每批次中得到最大为 size \* 主分片数 个文档。

JAVA示例

public class ScrollTest {
    	public static void main(String[] args) {
    		JSONObject resultObject = null;
    		Client esClient = ESClientHelper.getInstance().getClient();
    		SearchResponse searchResponse = esClient.prepareSearch("index")
    				.setSearchType(SearchType.SCAN)
    				// 实际返回的数量为5*index的主分片个数
    				.setSize(5)
    				// 这个游标维持多长时间
    				.setScroll(TimeValue.timeValueMinutes(8)).execute().actionGet();
    		// 第一次查询，只返回数量和一个scrollId
    		System.out.println(searchResponse.getScrollId());
    		System.out.println(searchResponse.getHits().getTotalHits());
    		System.out.println(searchResponse.getHits().hits().length);
    		System.out.println("------------------------------");
    		// 使用上次的scrollId继续访问
    		ScrollTest scroll = new ScrollTest();
    		do{
    			int num = scroll.scanData(esClient,searchResponse.getScrollId());
    			if(num ==0) break;
    		}while(true);
    		System.out.println("------------------------------END");
    	}
    	
    	private int scanData (Client esClient, String scrollId){
    		SearchResponse searchResponse = esClient.prepareSearchScroll(scrollId)
    				.setScroll(TimeValue.timeValueMinutes(8)).execute().actionGet();
    		System.out.println(searchResponse.getScrollId());
    		System.out.println(searchResponse.getHits().getTotalHits());
    		int num = searchResponse.getHits().hits().length;
    		System.out.println(searchResponse.getHits().hits().length);
    		JSONObject resultObject = null;
    		for (SearchHit hit : searchResponse.getHits()) {
    			String json = hit.getSourceAsString();
    			try {
    				resultObject = new JSONObject(json);
    			} catch (JSONException e) {
    				e.printStackTrace();
    			}
    		}
    		return num;
    	}
    }