OpenSearch踩坑记录

Posted 2024-11-14 Updated 2024-11- 14

By 十六

14~18 min read

Opensearch继承这里不做赘述，文中代码皆为Demo，不可直接用于生产需要包装。

分页查询相关

Opensearch是支持分页查询功能的其代码如下：

import org.opensearch.client.RequestOptions;
import org.opensearch.client.RestHighLevelClient;
import org.opensearch.client.indices.GetIndexRequest;
import org.opensearch.client.indices.GetIndexResponse;
import org.opensearch.index.query.QueryBuilders;
import org.opensearch.search.builder.SearchSourceBuilder;
import org.opensearch.client.RequestOptions;
import org.opensearch.search.SearchRequest;
import org.opensearch.search.SearchResponse;
import org.opensearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.opensearch.search.aggregations.AggregationBuilders;

import java.io.IOException;

public class OpenSearchPaginationClient extends RestHighLevelClient {

    public OpenSearchPaginationClient(RestClientBuilder builder) {
        super(builder);
    }

    public SearchResponse searchWithPagination(String index, int page, int size) throws IOException {
        // 创建搜索请求
        SearchRequest searchRequest = new SearchRequest(index);
        
        // 构建搜索源
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        
        // 设置查询条件（示例为匹配所有）
        searchSourceBuilder.query(QueryBuilders.matchAllQuery());
        
        // 设置分页参数
        searchSourceBuilder.from(page * size); // 从第几条记录开始
        searchSourceBuilder.size(size); // 每页的条目数
        
        // 将源构建器应用到搜索请求
        searchRequest.source(searchSourceBuilder);
        
        // 执行搜索并返回响应
        return this.search(searchRequest, RequestOptions.DEFAULT);
    }

    public static void main(String[] args) {
        // 需要替换为您的 OpenSearch 客户端构建逻辑
        RestHighLevelClient client = new OpenSearchPaginationClient(/* Your RestClientBuilder */);
        
        int page = 0; // 目标页码
        int size = 10; // 每页文档数量
        
        try {
            SearchResponse response = client.searchWithPagination("your_index_name", page, size);
            // 处理搜索响应，如打印搜索结果
            Arrays.stream(response.getHits().getHits()).forEach(hit -> {
                System.out.println("Document ID: " + hit.getId());
                System.out.println("Source: " + hit.getSourceAsString());
            });
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                client.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

在使用传统的基于 from 和 size 的分页方法时，尤其是在数据量较大的情况下，可能会遇到以下几个问题：

性能问题 : 随着页码的增加，from 参数的值也在增加，这会导致 OpenSearch 在执行查询时，跳过越来越多的文档，从而导致性能下降。
重复数据 : 如果数据在分页之间发生了变化（例如插入、更新或删除），可能会导致某些文档在不同的页面上重复出现，或者某些文档在某个页面上缺失。

为了避免这些问题，建议使用游标分页 （或者称为深度分页），这种方式通过保持状态来避免使用 from 和 size 方法，以下列举几种常见的游标分页方法：

1. 使用 Search After

search_after 是 OpenSearch 和 Elasticsearch 提供的一个高效的分页方法。它可以根据最后一条文档的信息进行下一页的查询，避免了传统分页方法的性能问题。

public SearchResponse searchWithSearchAfter(String index, int size, Object[] searchAfter) throws IOException {
    SearchRequest searchRequest = new SearchRequest(index);
    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();

    // 设置查询条件（示例为匹配所有）
    searchSourceBuilder.query(QueryBuilders.matchAllQuery());
    
    // 设置每页的条目数
    searchSourceBuilder.size(size); 

    // 使用 search_after 实现游标分页
    if (searchAfter != null) {
        searchSourceBuilder.searchAfter(searchAfter);
    }

    searchRequest.source(searchSourceBuilder);

    return this.search(searchRequest, RequestOptions.DEFAULT);
}

在该方法中，您可以传入 searchAfter 参数（一个 Object 数组），它表示上一个页面的最后一条文档的排序值。

改善数据的一致性

结合 search_after 使用时，可以确保在数据变更时，有效地减少重复或者遗漏。例如：

在文档或索引上使用时间戳字段，以确保在分页时可以通过自然排序来检索数据。
使用唯一标识符组合（如时间戳 + ID）作为 search_after 的参数。

2. 使用 Scroll API

如果需要在大数据集上实施深度分页，Scroll API 也是一个优秀的方案。Scroll API 主要用于查找所有文档，特别是在搜索结果集非常大的情况下可以减少性能开销。

public SearchResponse searchWithScroll(String index, int size) throws IOException {
    // 创建首次搜索请求
    SearchRequest searchRequest = new SearchRequest(index);
    searchRequest.scroll(TimeValue.timeValueMinutes(1)); // 设置 Scroll 的时间
    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
    searchSourceBuilder.query(QueryBuilders.matchAllQuery());
    searchSourceBuilder.size(size);
    searchRequest.source(searchSourceBuilder);

    // 执行首次搜索
    SearchResponse searchResponse = this.search(searchRequest, RequestOptions.DEFAULT);
    
    // 在这里处理首次搜索的结果
    // ...

    String scrollId = searchResponse.getScrollId();
    while (searchResponse.getHits().getHits().length != 0) {
        // 处理数据...
        
        // 创建新的 Scroll 请求
        SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
        scrollRequest.scroll(TimeValue.timeValueMinutes(1));

        // 执行 Scroll 请求
        searchResponse = this.scroll(scrollRequest, RequestOptions.DEFAULT);
        scrollId = searchResponse.getScrollId();
    }

    // 清除 Scroll 上下文
    ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
    clearScrollRequest.addScrollId(scrollId);
    this.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);

    return searchResponse;
}

注意事项：

使用 search_after : 当你的数据支持用来排序的字段时，这是处理深度分页的最佳解决方案。
使用 Scroll API : 如果有复杂的场景需要遍历大量文档，使用 Scroll API 是合适的选择。
对于实时性要求较高的场景，需在实现中充分考虑数据一致性和状态管理。

高手之路

Java

License: CC BY 4.0