[ElasticSearch] Scroll API를 사용하여 검색 / JAVA, Python 코드

[ElasticSearch] Scroll API를 사용하여 검색 / JAVA, Python 코드

 

이전 포스팅에 작성한 대로, 엘라스틱서치에서 1만 건 이상 대량의 데이터를 한 번에 검색하려고 하면 오류가 발생한다.

따라서, 검색조건을 잘 설정하여 1만 건 이하로 검색하거나 max_result_window를 늘려 검색하는 방법이 있다.

 

그러나 무작정 max_result_window를 늘리게 되면, 많은 리소스 사용으로 성능 문제를 야기할 수 있으며,

최대로 늘릴 수 있는 값이 5만건으로 한정적이기 때문에 권장되지 않는다.

 

👉관련글 보기 : [👩‍💻TIL/엘라스틱서치] QueryPhaseExecutionException: Result window is too large 에러

 

그래서 내가 대량 검색에 사용한 방법은 Scroll API를 사용하는 것이었다.

💡Scroll API란?

엘라스틱서치에서 기본적인 검색(search API)은 한 페이지를 리턴하고 나면 search context가 소멸된다. search API에서 from, to 값을 이용해 pagination을 구현한 것은 별개의 쿼리가 매번 수행된 것으로 RDBMS로 치면 cursor가 소멸된 것과 같다.  

 

대량의 데이터를 조회하기 위해서는 기존의 검색과 다른 방식을 사용해야 하는데,
엘라스틱서치에서 RDBMS의 커서(cursor)와 같은 기능을 하는 것이 바로 Scroll API이다.
(scroll 외에 Search_after 기능을 사용할 수도 있다.)

화면에서 스크롤을 내리면 아래쪽의 페이지의 데이터를 가져오듯,
정해진 스크롤 시간마다 검색 쿼리를 반복하여 그다음에 해당하는 데이터를 가져오는 방식의 검색을 말한다. 
스크롤은 매우 큰 결과 집합을 반복하는 것이나 색인 재생성(re-indexing) 작업에 유용하게 사용된다. 

💡사용 방법

curl을 이용해 간단히 scroll API를 사용해보면,

먼저 기본적인 search 쿼리에 scroll argument만 추가해서 쿼리 하면 된다. scroll 값이 추가되게 되면 elasticsearch는 앞으로 지정한 시간만큼은 지금 수행한 쿼리의 search context를 유지시켜준다.

예를 들어, student 인덱스에 저장된 모든 Issue Number 값을 조회한다고 했을 때,

다음과 같이 기본적인 match_all 쿼리에 scroll=10m 파라미터만 추가해서 요청한다.

 

localhost:9200/student/_search?scroll=10m -d'  


{  
  _source:[Issue Number],  
   size: 100,  
  query: { match_all : {}  
    }  
 } 

💡JAVA 코드 예시

curl을 사용하지 않고 자바로 엘라스틱서치의 scroll API를 구현하는 예시는 아래와 같다.

 

import static org.elasticsearch.index.query.QueryBuilders.*;


QueryBuilder qb = termQuery("multi", "test"); //검색할 쿼리문을 쿼리빌더로 작성해준다.

SearchResponse scrollResp = client.prepareSearch(test) 
        .addSort(FieldSortBuilder.DOC_FIELD_NAME, SortOrder.ASC) //정렬기준을 정해준다.
        .setScroll(new TimeValue(60000)) //스크롤 시간을 정해준다.
        .setQuery(qb) // 작성한 쿼리문 넣어준다.
        .setSize(100).get(); //각 스크롤 마다 100개의 hit을 받아온다.
        
// 더이상 받아올 hits가 없을 때 까지 Scroll반복한다.
do {
    for (SearchHit hit : scrollResp.getHits().getHits()) {
      // hit 데이터 처리해주는 부분
    }

    scrollResp = client.prepareSearchScroll(scrollResp.getScrollId()).setScroll(new TimeValue(60000))
                .execute().actionGet();
                
} while(scrollResp.getHits().getHits().length != 0); // hit가 더이상 없을 때 while문을 빠져나간다. 

 

자바 코드를 살펴보면 반복문을 통해 hit를 가져오며, 더 이상 가져올 것이 없을 때 스크롤 검색이 종료되는 것을 볼 수 있다.

 

💡 Python 코드 예시

 

search API를 먼저 호출할 때 scroll 속성을 추가해주고, 이어서 나머지 부분을 조회할 때에는 Elasticsearch.scoll 메서드를 이용한다. 사용 방법은 앞선 curl의 방법과 같다.

from elasticsearch import Elasticsearch
_KEEP_ALIVE_LIMIT='30s'
body = { 
  _source:[Issue Number],
  query : { 
    match_all:{}
  }
}

es_client = Elasticsearch([localhost:9200],timeout=300)
response = es_client.search(
  index = 'student',
  doc_type = 'doc',
  scroll = _KEEP_ALIVE_LIMIT,
  size = 100,
  body = body
  )

sid = response['_scroll_id']
fetched = len(response['hits']['hits'])

nums = []
for i in range(fetched):
  nums.append(int(response['hits']['hits'][i]['_source']['Issue Number']))

while(fetched>0):
  response = es_client.scroll(scroll_id=sid, scroll=_KEEP_ALIVE_LIMIT)
  fetched = len(response['hits']['hits'])
  for i in range(fetched):
    nums.append(int(response['hits']['hits'][i]['_source']['Issue Number']))

print(nums)

 

 

참고

https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/java-search-scrolling.html

Using scrolls in Java | Java API [7.6] | Elastic

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

https://oboki.net/workspace/bigdata/elasticsearch/scroll-api/