Elasticsearch 스크롤 조회 + 스레드 탱크 처리 대량의 데이터

스크롤 조회란 무엇입니까?

scroll 조회는 Elasticsearch에 대해 대량의 문서 조회를 효과적으로 실행할 수 있으며, 페이지를 깊이 나누는 대가를 치르지 않아도 된다.scroll 조회는 우리가 먼저 조회 초기화를 한 후에 결과를 대량으로 추출할 수 있도록 합니다.이것은 약간 전통 데이터베이스에 있는 커서 (cursor) 와 같다.
scroll 조회는 첫 번째 조회 때 당시 시간점의 스냅샷 데이터를 저장한 다음에 이 오래된 보기 스냅샷을 바탕으로 데이터 검색을 제공합니다. 이 기간에 데이터가 변경되면 이 스냅샷 데이터는 변하지 않습니다.
깊이 페이지를 나누는 대가의 근원은 결과집 전역 정렬이다. 전역 정렬의 특성을 없애면 검색 결과의 원가가 매우 낮다.scroll 조회는 _ 기반doc가 정렬하는 방식은 성능이 비교적 높다.
매번 scroll 요청을 보낼 때마다, 우리는 scroll Id를 지정해서 실효 시간을 설정해야 합니다. 매번 검색 요청은 이 실효 시간 안에 완성하면 됩니다.추가 스크롤 질의는 다음을 참조하십시오.https://www.elastic.co/guide/cn/elasticsearch/guide/current/scroll.html

java API의 스크롤 조회

MatchQueryBuilder qb = QueryBuilders.matchQuery("message"," ");

SearchResponse scrollResp = client.prepareSearch()
		.addSort(FieldSortBuilder.DOC_FIELD_NAME,SortOrder.ASC)
		.setScroll(new TimeValue(10))
		.setQuery(qb)
		.setSize(100).get();//  hits 100


//  ， hits ， hits 
while(scrollResp.getHits().getHits().length !=0) {
	for (SearchHit hit : scrollResp.getHits().getHits()) {
		System.out.println(hit.toString());
		// Handle the hit...
	}
	
	scrollResp = client.prepareSearchScroll(scrollResp.getScrollId()).setScroll(new TimeValue(100)).execute().actionGet();
}; 

elasticsearchTemplate를 사용하여 스크롤 조회

scrollId 가져오기:

String scrollId = elasticsearchTemplate.scan(searchQuery, 6000, false);

스크롤 질의:

Page pages = elasticsearchTemplate.scroll(scrollId, 6000, Entity.class);

스크롤 조회 + 스레드 풀

elasticsearchTemplate 주입:

@Autowired
private ElasticsearchTemplate elasticsearchTemplate;
//  ， 10 
private static ExecutorService executorService = Executors.newFixedThreadPool(10);

구현 코드:

String scrollId = null;
try {
	//  
	int SCROLL_TIMEOUT_INMILLIS = 12 * 60 * 60 * 1000;
	
	//  
	QueryBuilder orgQuery = QueryBuilders
			.boolQuery()
			.filter(QueryBuilders.termQuery("orgCode", orgCode));
	// es 
	SearchQuery searchQuery = new NativeSearchQueryBuilder()
			.withIndices(" index")
			.withTypes(" type")
			.withQuery(orgQuery)
			.withPageable(new PageRequest(0, 100)) //  100 
			.build();
				
	//  ID
	//  ， 
	scrollId = elasticsearchTemplate.scan(searchQuery, SCROLL_TIMEOUT_INMILLIS, false);
	while(true) {
		
		//  
		Page<DemoEntity> pages = elasticsearchTemplate.scroll(scrollId, SCROLL_TIMEOUT_INMILLIS, DemoEntity.class);
		
		//  ， while
		if(!pages.hasContent()) {
			break;
		}
		
		for(DemoEntity document:pages.getContent()) {
			//  
			executorService.execute(new clearMethod(document));
		}
	}
} catch (Exception e) {
	// exception
}finally {
	//  id
	elasticsearchTemplate.clearScroll(scrollId);
	//  
	executorService.shutdown();
}

class clearMethod extends Thread {
	private DemoEntity document;
	
	public clearMethod(DemoEntity document) {
		this.document = document;
	}
	public void run() {
		// do something...
    }
}

끝내다

만약elasticsearch에 대해 잘 모르면 저의 몇 편의 블로그를 참고할 수 있습니다.작업 요구 때문에 본 논문에서 기술한 코드 ES 버전은 2.4.0으로 유행이 지났을 수도 있지만 사고방식은 대체적으로 정확하다.본 편에서 보여준 코드는 위조 코드입니다. 직접copy를 하지 마십시오. 설명이 명확하지 않은 곳이 있으면 메시지를 남겨 주십시오.