背景介紹

AWS OpenSearch是AWS的一個檢索分析服務，是基于開源的Elasticsearch 7.x分支fork出來的獨立的一個代碼倉庫，做了獨立的維護，加入了一些自己的優化，本文在這里主要介紹是常見的基礎用法

引入相關依賴

 <dependency><groupId>org.opensearch.client</groupId><artifactId>opensearch-java</artifactId><version>2.17.0</version></dependency>

查詢返回指定屬性字段

按照前端要求的返回字段（“productId”, “title”, “rating”, “images”,“productTags”）進行返回，而不是返回所有字段

SearchResponse<ProductVO> search = openSearchClient.search(s -> s.index("product_index").source(c -> c.filter(e -> e.includes("productId", "title", "rating",  "images","productTags"))).query(q -> q.terms(t -> t.field("productId").terms(ts -> ts.value(values)))),ProductVO.class);

分頁查詢返回

根據前端傳入的分頁參數，當前頁(pageNo)和每頁的條數(pageSize)執行查詢

  PageResult<ProductVO> pageResult = new PageResult<>();Integer pageIndex = requestVO.getPageNo();int pageSize = requestVO.getPageSize() != null ? requestVO.getPageSize() : 10;int from = (pageIndex - 1) * pageSize;
SearchResponse<ProductVO> search = openSearchClient.search(s -> s.index("product_index").source(c -> c.filter(e -> e.includes("productId", "title", "rating",  "images","productTags"))).from(from).size(pageSize).query(q -> q.terms(t -> t.field("productId").terms(ts -> ts.value(values)))),ProductVO.class);
List<ProductVO> productList = Lists.newArrayList();                    
if (CollectionUtils.isNotEmpty(search.hits().hits())) {search.hits().hits().forEach(h -> {ProductVO productVo = h.source();productList.add(productVo);});}int total = Math.toIntExact(search.hits().total().value());pageResult.setCurrentPage(pageIndex);pageResult.setPageSize(pageSize);pageResult.setTotal(total);if (pageResult.getTotal() % pageResult.getPageSize() == 0) {pageResult.setTotalPage(pageResult.getTotal() / pageResult.getPageSize());} else {pageResult.setTotalPage((pageResult.getTotal() / pageResult.getPageSize()) + 1);}pageResult.setItems(productList);

復合查詢

List<Query> mustQueryList = new ArrayList<>();
List<Query> mustNotQueryList = new ArrayList<>();List<FieldValue> values = new ArrayList<>();
List<String> languages = List.of("zh");
languages.forEach(c -> {values.add(FieldValue.of(c));
});
int rating = Integer.parseInt("4.5");
Query ratingQuery = RangeQuery.of(r -> r.field("rating").gte(JsonData.of(rating - 0.25)).lt(JsonData.of(rating + 0.75))).toQuery();List<FieldValue> categoriesValues = new ArrayList<>();
categoriesValues.add(FieldValue.of("CA0001","CA0002"));
Query nestedQuery = NestedQuery.of(n -> n.path("categories").query(q -> q.terms(r -> r.field("categories.id").terms(t -> t.value(categoriesValues)).boost(1000f)))
).toQuery();
mustQueryList.add(TermsQuery.of(t -> t.field("language").terms(new TermsQueryField.Builder().value(values).build())).toQuery());
mustQueryList.add(ratingQuery);
mustNotQueryList.add(nestedQuery);
Query complexQuery = BoolQuery.of(b -> b.must(mustQueryList).mustNot(mustNotQueryList)).toQuery();
SearchResponse<ProductVO> search = openSearchClient.search(s -> s.index("product_index").source(c -> c.filter(e -> e.includes("productId", "title", "rating",  "images","productTags"))).from(from).size(pageSize).query(complexQuery),ProductVO.class);

聚合統計

以下是聚合查詢商品每個評分的數量

 Aggregation aggregation = Aggregation.of(a -> a.terms(ts -> ts.field("rating").size(1000)));
SearchResponse<ProductBO> search = openSearchClient.search(s -> s.index("product_index").source(c -> c.filter(e -> e.includes("productId", "title", "rating"))).aggregations("ratingAgg", aggregation).query(filterQuery),ProductBO.class);
if (null != search.aggregations()) {Collection<Aggregate> aggregateCollection = search.aggregations().values();List<FacetHit> facetHitsList = Lists.newArrayList();aggregateCollection.forEach(aggregate -> {String kind = aggregate._kind().name();log.info("The aggregation type is {}", kind);switch (kind) {case "Nested" -> {Collection<Aggregate> nestedAggregateCollection = aggregate.nested().aggregations().values();nestedAggregateCollection.forEach(nestedAggregate -> {addFacetHits(nestedAggregate, facetHitsList);});}case "Sterms" -> {addFacetHits(aggregate, facetHitsList);}case "Dterms" -> {Buckets<DoubleTermsBucket> buckets = aggregate.dterms().buckets();buckets.array().forEach(bucket -> {String key = String.valueOf(bucket.key());FacetHit facetHit = new FacetHit();facetHit.setCount(Math.toIntExact(bucket.docCount()));facetHit.setValue(key);facetHitsList.add(facetHit);});}default -> log.warn("Unrecognized type:{} cannot be processed", kind);}});
}private void addFacetHits(Aggregate aggregate, List<FacetHit> facetHitsList) {Buckets<StringTermsBucket> buckets = aggregate.sterms().buckets();List<StringTermsBucket> stringTermsBuckets = buckets.array();stringTermsBuckets.forEach(s -> {String key = s.key();long docCount = s.docCount();FacetHit facetHit = new FacetHit();facetHit.setCount(Math.toIntExact(docCount));Aggregate parentAggregate = s.aggregations().get("parent_docs");if (null != parentAggregate && AggregateConstants.KIND_REVERSE_NESTED.equals(parentAggregate._kind().name())) {ReverseNestedAggregate reverseNested = parentAggregate.reverseNested();if (reverseNested != null) {long parentDocCount = reverseNested.docCount();facetHit.setCount(Math.toIntExact(parentDocCount));}}facetHit.setValue(key);facetHitsList.add(facetHit);});
}

基礎排序

按照定義的排序字段進行排序

SearchResponse<ProductVO> search = openSearchClient.search(s -> s.index("product_index").source(c -> c.filter(e -> e.includes("productId", "title", "rating",  "images","productTags"))).from(from).size(pageSize).query(q -> q.terms(t -> t.field("productId").terms(ts -> ts.value(values)))).sort(t -> t.field(f -> f.field("publishDate").order(SortOrder.Desc))) .sort(t -> t.field(f -> f.field("rating").order(SortOrder.Desc))),ProductVO.class);

高階排序

按照特定的一批商品排在查詢結果的最前面

List<String> topProductIdList = List.of("1","2");//特定的商品編號
SearchResponse<ProductVO> search = openSearchClient.search(s -> s.index("product_index").source(c -> c.filter(e -> e.includes("productId", "title", "rating",  "images","productTags"))).from(from).size(pageSize).query(q -> q.terms(t -> t.field("productId").terms(ts -> ts.value(values)))).sort(getHightSortOptions(topProductIdList)), ProductVO.class);private List<SortOptions> getHightSortOptions(List<String> topProductIdList) {List<SortOptions> sortOptions = Lists.newArrayList();sortOptions.add(SortOptions.of(f -> f.script(st -> st.type(ScriptSortType.Number).script(Script.of(sf -> sf.inline(ie -> ie.source("params.topProductIds.indexOf(doc['productId'].value) >= 0 ? params.topProductIds.indexOf(doc['productId'].value) : params.topProductIds.size()").lang("painless").params(Map.of("topProductIds", JsonData.of(topProductIdList)))))).order(SortOrder.Asc))));sortOptions.add(SortOptions.of(t -> t.field(f -> f.field("rating").order(SortOrder.Desc))));sortOptions.add(SortOptions.of(t -> t.field(f -> f.field("publishDate").order(SortOrder.Desc))));return sortOptions;}

查看OpenSearch的數據

可以通過OpenSearch dashboard查看，如下圖所示：