从ES数据库中随机抽样10%的数据

Posted 2022-12-21 Lana啦啦啦

如果数据量比较小，从 Elasticsearch 数据库中随机抽取 10% 的数据，可以使用以下步骤来实现：

1. 使用 Elasticsearch 的 Search API 搜索数据库中的数据。
2. 使用 QueryBuilder 构建查询语句，并指定要查询的索引和类型。
3. 在查询中添加一个随机排序的子句，例如：

queryBuilder.addSort(new FieldSortBuilder("_doc").order(SortOrder.ASC).sortMode(SortMode.RANDOM));

4. 使用 SearchResponse 执行查询，并使用 SearchHits 获取查询结果。
5. 使用 SearchHits 的 getTotalHits 方法计算总共有多少文档符合查询条件。
6. 计算出 10% 的数量，并使用 SearchHits 的 getHits 方法获取查询结果，只取前 10% 个文档即可。
   以下是一个示例方法，该方法可以实现上述步骤：

public List<Map<String, Object>> sampleData(String index, String type, int sampleSize) 
    // 1. 使用 Elasticsearch 的 Search API 搜索数据库中的数据
    SearchRequest searchRequest = new SearchRequest(index);
    searchRequest.types(type);

    // 2. 使用 QueryBuilder 构建查询语句，并指定要查询的索引和类型
    QueryBuilder queryBuilder = QueryBuilders.matchAllQuery();

    // 3. 在查询中添加一个随机排序的子句
    queryBuilder.addSort(new FieldSortBuilder("_doc").order(SortOrder.ASC).sortMode(SortMode.RANDOM));

    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
    searchSourceBuilder.query(queryBuilder);
    searchRequest.source(searchSourceBuilder);

    // 4. 使用 SearchResponse获取前10%的数据
    //todo

如何处理大量数据的情况

如果要处理的数据量很大，可能会对 Elasticsearch 的性能造成压力。 为了解决这个问题，可以使用 Elasticsearch 的 Scroll API。

Scroll API 可以用来执行批量查询，它会在内部使用滚动查询，允许在一个查询中搜索大量的数据，并将结果分批返回。这样，您就可以在单个查询中处理大量的数据，而不会对 Elasticsearch 造成太大的压力。

下面是一个示例方法，该方法使用 Scroll API 来实现随机抽样：

public List<Map<String, Object>> sampleDataWithScroll(String index, String type, int sampleSize) 
    List<Map<String, Object>> result = new ArrayList<>();

    // 1. 使用 Elasticsearch 的 Search API 搜索数据库中的数据
    SearchRequest searchRequest = new SearchRequest(index);
    searchRequest.types(type);

    // 2. 使用 QueryBuilder 构建查询语句，并指定要查询的索引和类型
    QueryBuilder queryBuilder = QueryBuilders.matchAllQuery();

    // 3. 在查询中添加一个随机排序的子句
    queryBuilder.addSort(new FieldSortBuilder("_doc").order(SortOrder.ASC).sortMode(SortMode.RANDOM));

    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
    searchSourceBuilder.query(queryBuilder);
    searchRequest.source(searchSourceBuilder);

    // 4. 使用 SearchResponse 执行查询
    SearchResponse searchResponse = client.search(searchRequest);

    // 5. 使用 SearchHits 获取查询结果
    SearchHits hits = searchResponse.getHits();

    // 6. 使用 SearchHits 的 getTotalHits 方法计算总共有多少文档符合查询条件
    long totalHits = hits.getTotalHits();

    // 7. 计算出 10% 的数量
    int sampleCount = (int) (totalHits * sampleSize);

    // 8. 设置滚动参数
    String scrollId = searchResponse.getScrollId();
    searchRequest.scroll(TimeValue.timeValueMinutes(1));

    while (result.size() < sampleCount) 
	    searchResponse = client.searchScroll(new SearchScrollRequest(scrollId).scroll(TimeValue.timeValueMinutes(1)));
	    hits = searchResponse.getHits();
	
	    // 9. 使用 SearchHits 的 getHits 方法获取查询结果，只取前 10% 个文档
	    for (int i = 0; i < hits.getHits().length && result.size() < sampleCount; i++) 
	        result.add(hits.getAt(i).getSourceAsMap());
	    
	
	    // 10. 更新滚动 ID
	    scrollId = searchResponse.getScrollId();