在es中用scroll查询与completableFuture

Posted 2021-01-20 029zz010buct

一般而言，es返回数据的上限是10000条，如果超过这个数量，就必须使用scroll查询。

所谓scroll查询就类似DBMS中的游标，或者快照吧，利用查询条件，在第一次查询时，在所有的结果上形成了一个快照，然后再分批分次的读取出来。

要完成一个scroll查询分两个阶段：

阶段一：带查询参数

1 POST /twitter/_search?scroll=1m
2 {
3     "size": 100,
4     "query": {
5         "match" : {
6             "title" : "elasticsearch"
7         }
8     }
9 }

这个查询条件比较简单，只是示意。

关键是有两点：1.post路径中的scroll关键字，指明是一个scroll查询；2，scroll=1m意味着查询结果数据在es的服务器有效期是一分钟。

在查询结果的返回值中会带有一个scroll id的参数，这个参数在第二次查询的时候需要。

阶段二：不带参数查询

POST  /_search/scroll 
{
    "scroll" : "1m", 
    "scroll_id" : "DXF1ZXJ5QW5kRmV0Y2gBAAAAAAAAAD4WYm9laVYtZndUQlNsdDcwakFMNjU1QQ==" 
}

这个查询请求中，会带上第一次请求得到的scroll_id这个字段。

然后循环往复，第三次查询需要带上第二次查询返回的scroll_id，以此类推，就这个例子而言，当判断返回的数据条数小于100条的时候，就可以结束请求了。

使用scroll查询的两个优势是：

1.无论查询的数据量是多大，都能够查询成功。

2.准确反映了第一次查询当时的查询结果，第一次查询之后的查询请求不会包含新的数据。

但也有一个缺点：

1.因为查询的递进的，第二次依赖于第一次，第三次依赖于第二次，所以如果数据量很多，查询的耗时就比较长。

如何解决耗时长这个问题了？就不能使用scroll来查询了，使用常规的查询，但是启用多线程去查。

GET /_search
{
    "query" : {
        "term" : { "user" : "kimchy" }
    }
}

加入在常规的查询中有timestamp这样的自动，我们可以预先对timestamp进行划分，比如分出10份，当然前提是我们假设数据在时间上是均匀的，然后每个时间切分启用一个线程去查询。在java中有completableFuture能够比较好的支持这种查询场景。

CompletableFuture<JSONArray>[] futures = (CompletableFuture<JSONArray>[]) new CompletableFuture[count];
                for (int i = 0; i < count; i++) {
                    CompletableFuture<JSONArray> future = CompletableFuture.
                            supplyAsync(new JSONAarrySupplier(this.queryString,timestamp[i])
                            .exceptionally(ex -> {
                                logger.error(ex.getMessage());
                                return null;});
                    futures[i] = future;
                }
                CompletableFuture<List<JSONArray>> allFuture = myAllOf(futures);
                result = allFuture.get();

如上述，在一开始建立了一个future数组，然后根据时间切片，构建查询请求，并放入completableFuture中。

在最后调用get方法，拿到所有线程执行完的结果。

这里有一个点要注意，就是completableFuture.allOf方法本身返回的是void，如果我们的future是有返回值的话，就不能直接调用java自身提供的，需要改下一下，如上其实调用了下面的方法：

 public static CompletableFuture<List<JSONArray>> myAllOf(CompletableFuture<?>... futures) {
        return CompletableFuture.allOf(futures)
                .thenApply(x -> Arrays.stream(futures)
                        .map(f -> (JSONArray) f.join())
                        .collect(toList())
                ).exceptionally(ex -> {
                    logger.error(ex.getMessage());
                    return null;});
    }

这个方法中实现了返回值的转换。

这种多线程的查询，相对于scroll去查询，在网络不是瓶颈的前提下，性能还是有很大提升。

 

综上所述：

1.如果对时间不敏感，还是推荐使用scroll查询，毕竟反映了查询时间点的实际情况。

2.如果对时间敏感，则需要合理挑选查询分片条件，形成合理的多线程查询。

参考https://www.elastic.co/guide/en/elasticsearch/reference/current/search-request-scroll.html