엘라스틱 서치 인덱스를 생성하고 스프링에서 질의하기

웹크롤링 데이터가 워낙 방대하다보니 RDB를 사용하면 질의 시 수초에서 수십초까지 기다려야하는 상황이 발생하여 검색엔진인 엘라스틱 서치를 도입한 과정을 보여드리려 합니다.

1. 엘라스틱 서치 인덱스 생성

엘라스틱 서치를 AWS 서버에 설치한 후 실행시키고 kibana 또한 설치하여 엘라스틱 서치에 kibana를 이용해 접근 후 인덱스를 아래와 같이 생성하였습니다.

PUT goods
{
  "settings": {
    "index.max_result_window": 800,
    "index": {
      "analysis": {
        "tokenizer": {
          "my_nori_tokenizer": {
            "type": "nori_tokenizer",
            "decompound_mode": "mixed",
            "discard_punctuation": "false"
          },
          "my_ngram_tokenizer": {
            "type": "ngram",
            "min_gram": 2,
            "max_gram": 3
          }
        },
        "filter": {
          "stopwords": {
            "type": "stop",
            "stopwords": [" "]
          }
        },
        "analyzer": {
          "my_nori_analyzer": {
            "type": "custom",
            "tokenizer": "my_nori_tokenizer",
            "filter": ["lowercase", "stop", "trim", "stopwords", "nori_part_of_speech"],
            "char_filter": ["html_strip"]
          },
          "my_ngram_analyzer": {
            "type": "custom",
            "tokenizer": "my_ngram_tokenizer",
            "filter": ["lowercase", "stop", "trim", "stopwords", "nori_part_of_speech"],
            "char_filter": ["html_strip"]
          }
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "name": {
        "type": "text",
        "analyzer": "standard",
        "search_analyzer": "standard",
        "fields": {
          "nori": {
            "type": "text",
            "analyzer": "my_nori_analyzer",
            "search_analyzer": "my_nori_analyzer"
          },
          "ngram": {
            "type": "text",
            "analyzer": "my_ngram_analyzer",
            "search_analyzer": "my_ngram_analyzer"
          },
          "keyword": {
            "type": "keyword",
            "ignore_above": 256
          }
        }
      },
      "@timestamp": {
        "type": "date"
      },
      "detail": {
        "type": "text",
        "fields": {
          "keyword": {
            "type": "keyword",
            "ignore_above": 256
          }
        }
      },
      "image": {
        "type": "text",
        "fields": {
          "keyword": {
            "type": "keyword",
            "ignore_above": 256
          }
        }
      },
      "insertion_time": {
        "type": "date"
      },
      "is_deleted": {
        "type": "long"
      },
      "modification_time": {
        "type": "date"
      },
      "price": {
        "type": "long"
      },
      "sellid": {
        "type": "text",
        "fields": {
          "keyword": {
            "type": "keyword",
            "ignore_above": 256
          }
        }
      },
      "amount": {
        "type": "integer"
      },
      "discountamount": {
        "type": "integer"
      },
      "discountrate": {
        "type": "float"
      },
      "deliveryfee": {
        "type": "integer"
      }
    }
  }
}

위 인덱스를 보면

(1). "index.max_result_window": 800, 로 설정해 최대 800개 까지만 결과를 return하게끔 설정해서 과부하를 방지하고

(2). ES 분석기로 nori_tokenizer(한국어 형태소 분석기), ngram(문자열을 잘라서 토큰화하는 분석기)를 설정하고 

(3). "filter": ["lowercase", "stop", "trim", "stopwords", "nori_part_of_speech"],
            "char_filter": ["html_strip"]

로 분석기를 커스터마이징해서 lowercase(소문자로 대체해서 대소문자 구분 제거), stop/stopwords(검색에서 자주 제외되는 불용어 제거), trim(문자 앞뒤 공백 제거), nori_part_of_speech(형태소 분석정보 포함), html_strip(입력 텍스트에서 HTML 태그를 제거) 하는 설정을 부여합니다.

(4). 사용자가 각 상품의 name을 이용해 검색하므로 name 필드에 기본 standard analyzer, nori analyzer, ngram analyzer를 설정합니다.

 

2. 스프링에서 질의하기

(1) ElasticSearchConfig

@Configuration
public class ElasticsearchConfig {
	
	@Value("${elasticsearch.url}")
    private String url;
	
	@Value("${elasticsearch.port}")
    private String port;
	
    @Bean
    // NativeSearchQuery 을 이용해 요청보낼 때 사용
    public ElasticsearchOperations elasticsearchOperations() {
        ClientConfiguration clientConfiguration = ClientConfiguration.builder()
                .connectedTo(url + ":" + port)
                .build();

        return new ElasticsearchRestTemplate(RestClients.create(clientConfiguration).rest());
    }
    
    @Bean
    // SearchSourceBuilder 을 이용해 요청보낼 때 사용
    public RestHighLevelClient client() {
    	return new RestHighLevelClient(
    		    RestClient.builder(
    		        // Elasticsearch 호스트 및 포트 설정
    		        new HttpHost(url, Integer.parseInt(port), "http")
    		        // 추가적인 Elasticsearch 호스트나 포트가 있다면 여기에 추가
    		    )
    		);
    }
}

ElasticSearchOperations를 이용해 질의하는 방식과 RestHighLevelClient를 이용해 질의하는 방식을 구현하였습니다.

(2) ElasticSearchOperations를 이용해 질의하는 방식

public List<Goods> getDataFromElasticsearch(SearchDto params, String date) {
    	// BoolQueryBuilder 생성
    	BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();

    	if(params.getSearchValue() != null && !params.getSearchValue().equals("")) {
    		// 검색어 조건 포함
    		BoolQueryBuilder mustQuery = QueryBuilders.boolQuery();
            // Standard Analyzer
    		mustQuery.should(QueryBuilders.matchQuery("name", params.getSearchValue()));
            // Nori Analyzer
    		mustQuery.should(QueryBuilders.matchQuery("name.nori", params.getSearchValue()));
            // Ngram Analyzer
    		mustQuery.should(QueryBuilders.matchQuery("name.ngram", params.getSearchValue()));
    		boolQuery.must(mustQuery);
    	}
    	
    	// is_deleted 조건 포함
    	boolQuery.filter(QueryBuilders.termQuery("is_deleted", 0));
    	
    	// 가격 조건 포함
    	if(params.getSearchMinPrice() != null && params.getSearchMinPrice() > 0) {
    		boolQuery.filter(QueryBuilders.rangeQuery("price").gte(params.getSearchMinPrice()));
    	}
    	if(params.getSearchMaxPrice() != null && params.getSearchMaxPrice() > 0) {
    		boolQuery.filter(QueryBuilders.rangeQuery("price").lte(params.getSearchMaxPrice()));
    	}

    	// insertion_time 조건 포함
    	boolQuery.filter(QueryBuilders.matchQuery("insertion_time", date));
    	
    	// NativeSearchQuery를 사용하여 쿼리 실행
    	NativeSearchQueryBuilder searchQuery = new NativeSearchQueryBuilder()
    	        .withQuery(boolQuery)
    	    	// 최소 적합도 20으로 설정
    	        .withMinScore(20)
    	        // 페이징 처리
    	        .withPageable(PageRequest.of(params.getPage() - 1, params.getRecordSize()));
    	
    	// ORDER BY 절 추가
    	if (params.getOrderBy() == null || params.getOrderBy().equals("")) {
    		searchQuery.withSort(Sort.by(Sort.Direction.DESC, "_score"));
    	} else if (params.getOrderBy().equals("priceASC")) {
    		searchQuery.withSort(Sort.by(Sort.Direction.ASC, "price"));
    	} else if (params.getOrderBy().equals("priceDESC")) {
    		searchQuery.withSort(Sort.by(Sort.Direction.DESC, "price"));
    	} else if (params.getOrderBy().equals("dateASC")) {
    		searchQuery.withSort(Sort.by(Sort.Direction.ASC, "insertion_time"));
    	} else if (params.getOrderBy().equals("dateDESC")) {
    		searchQuery.withSort(Sort.by(Sort.Direction.DESC, "insertion_time"));
    	}

    	// NativeSearchQuery를 사용하여 쿼리 생성
    	NativeSearchQuery searchQueryComplete = searchQuery.build();
    	
    	logger.info("[ getDataFromElasticsearch ] Query : " + searchQueryComplete.getQuery().toString());
    	
        SearchHits<Goods> searchHits = elasticsearchOperations.search(searchQueryComplete, Goods.class);
        List<Goods> dataList = new ArrayList<>();
        for (SearchHit<Goods> searchHit : searchHits) {
            dataList.add(searchHit.getContent());
        }
        logger.info("[ getDataFromElasticsearch ] dataList.size() : " + dataList.size());
        return dataList;
    }

name을 이용한 검색어 질의뿐 아니라 is_deleted 필드가 0이어야된다는 조건, 특정 가격보다 크거나 작다는 조건, insert_time 필드가 date와 동일해야한다는 조건, 최소 적합도 설정, 페이징 처리, 정렬조건 등 여러 조건을 추가하고 NativeSearchQuery를 생성한 후 ElasticSearchOperations를 이용해 질의하고 있음을 확인할 수 있습니다.

(3) RestHighLevelClient를 이용해 질의하는 방식

public List<Statistic> getStatisticData(String search) throws IOException {
    	
    	// 검색어 조건 포함
        // Standard Analyzer
    	QueryBuilder nameMatchQueryBuilder = QueryBuilders.matchQuery("name", search)
    	        .operator(Operator.OR);
		// Nori Analyzer
    	QueryBuilder nameNoriMatchQueryBuilder = QueryBuilders.matchQuery("name.nori", search)
    	        .operator(Operator.OR);
		// Ngram Analyzer
    	QueryBuilder nameNgramMatchQueryBuilder = QueryBuilders.matchQuery("name.ngram", search)
    	        .operator(Operator.OR);

    	QueryBuilder mustQueryBuilder = QueryBuilders.boolQuery()
    	        .should(nameMatchQueryBuilder)
    	        .should(nameNoriMatchQueryBuilder)
    	        .should(nameNgramMatchQueryBuilder);

    	// is_deleted 조건 포함
    	QueryBuilder filterQueryBuilder = QueryBuilders.termQuery("is_deleted", 0);

    	// 일별로 집계해서 (price)의 평균값을 산출함. 이 때 일자를 DESC로 정렬함
    	DateHistogramInterval interval = DateHistogramInterval.DAY;
    	DateHistogramAggregationBuilder aggregations = AggregationBuilders.dateHistogram("dates")
    													.field("insertion_time")
    													.fixedInterval(interval)
    													.subAggregation(AggregationBuilders.avg("average_price").field("price"))
    													.order(BucketOrder.key(false));

    	SearchSourceBuilder query = SearchSourceBuilder.searchSource().query(QueryBuilders.boolQuery()
															                .must(mustQueryBuilder)
															                .filter(filterQueryBuilder)).aggregation(aggregations).size(0);

    	// 최소 적합도 20으로 설정
    	query.minScore(20.0f);
    	
    	String jsonQuery = query.toString();
    	logger.info("####### jsonQuery : {}", jsonQuery);
    	
    	SearchRequest request = new SearchRequest();
    	request.indices("goods").source(query);
    	SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    	
    	logger.info("####### response : {}", response.toString());
    	Aggregations aggregation = response.getAggregations();
    	
    	logger.info("####### Aggregations : {}", aggregation.toString());
    	List<Statistic> statistics = new ArrayList<Statistic>();
    	
    	if (aggregation != null) {
    	    ParsedDateHistogram dateHistogram = aggregation.get("dates");

    	    if (dateHistogram != null) {
    	        List<? extends Bucket> buckets = dateHistogram.getBuckets();
    	        int count = 0;
    	        for (Bucket bucket : buckets) {
    	        	count++;
    	        	
    	        	// 최신일자순으로(일자별 DESC정렬된 데이터) 30개까지만 return
    	        	if(count > 30) break;
    	        	
    	        	Statistic statistic = new Statistic();
    	            String keyAsString = bucket.getKeyAsString().substring(0, 10);
    	            long docCount = bucket.getDocCount();
    	            Avg averagePriceAggregation = bucket.getAggregations().get("average_price");
    	            double averagePrice = averagePriceAggregation.getValue();

    	            statistic.setKeyAsString(keyAsString);
    	            statistic.setDocCount(docCount);
    	            statistic.setAveragePrice((int) averagePrice);
    	            
    	            statistics.add(statistic);
    	        }
    	    }
    	}
    	return statistics;
    }

name을 이용한 검색어 질의뿐 아니라 is_deleted 필드가 0이어야된다는 조건, 일별로 집계해서 price의 평균값을 계산한다는 조건, 최소 적합도 등 여러 조건을 추가하고 SearchSourceBuilder를 생성한 후 RestHighLevelClient를 이용해 질의하고 있음을 확인할 수 있습니다.

 

이상입니다. 지금까지 읽어주셔서 감사합니다.