optimization - 使用ElasticSearch Scroll API时，如何现场优化时间参数？

Question

我正在使用elasticsearch滚动api返回大量文档。根据documentation，

“滚动终止时间在我们每次运行滚动请求时都会刷新，因此它只需要足够长的时间即可处理当前批次的结果，而不是所有与查询匹配的文档。超时非常重要，因为保持滚动窗口打开会消耗资源，我们希望在不再需要它们时立即释放它们。设置超时可使Elasticsearch在一段时间不 Activity 后自动释放资源。”

我的问题是如何优化时间参数？在某些情况下，我需要请求和处理约600页，但是在第300页上它会失败(很长的路要走!)。我怀疑是否可以优化传递的时间参数，这样在使用ES资源时效率会更高，而且不容易出错。此代码正在此处的集群上进行测试，但可能会移植到许多其他集群上，因此我希望优化time参数以使其适应集群。另外，我不想在ES群集上使用比我需要的其他资源更多的B / c用户，大概也可以使用它。

这是我的主意。在最初的滚动请求中，传递一个慷慨的时间参数，例如5m，然后设置返回第一页结果所需的时间。然后在第二个滚动请求上，向它传递一个时间参数，该参数比观察到的第一个请求所花的时间稍大。归纳地说，每个页面请求的时间都比以前观察到的页面的完成时间稍长。假设由于每个页面返回的文档数量相同(在我的情况下，文档的大小几乎相同)，因此返回该页面所需的时间与以前观察到的时间大致相同。这个假设成立吗？

是否有更智能的方式来调整时间参数？并且就此而言，size参数(在上面的想法中，size参数保持不变)。

Answer 1

好的，我做了一些数据分析，并凭经验发现了一些东西。对于许多不同的尺寸，我运行了10-20页的滚动API查询。对于固定大小，返回页面所花费的时间大约是高斯，下面给出了相应的方法。

means =  {1000: 6.0284869194030763,
 1500: 7.9487858772277828,
 2000: 12.139444923400879,
 2500: 18.494202852249146,
 3000: 22.169868159294129,
 3500: 28.091009926795959,
 4000: 36.068559408187866,
 5000: 53.229292035102844}

我的下一个想法是，这可能取决于是否在计算机上运行了其他查询，因此我进行了实验，一半的页面是来自ES的唯一请求，另一半是在运行第二个滚动查询时。时间似乎没有改变。

最后，因为时间取决于给定的ES配置和带宽等。我提出了这种解决方案。

为初始页面设置了慷慨的页面时间。

每页时间

在观察到的时间+一点和初始时间之间使用加权的运行平均值(因此您的时间参数始终比所需的要大一些，但会减少到平均值+一点)。这是一个例子:

尝试= 0size = 3000wait_time = 2 ##慷慨的开始时间
return_hits = {} ##页，匹配列表
而尝试<3:
尝试:
打印“\ n \ t以大小=％s ...运行警报滚动查询”“％(size)
页面= client.search(索引=索引，doc_type = doc_type，正文= q，滚动='1m'，search_type ='scan'，size = size)

    sid = page['_scroll_id'] ## scroll id
    total_hits = page['hits']['total'] ## how many results there are. 
    print "\t\t There are %s hits total." %(total_hits)

    p = 0 ## page count 
    doc_count = 0 ## document count 
    # Start scrolling
    while (scroll_size > 0):
        p += 1
        print "\t\t Scrolling to page %s ..." % p
        start = time.time()
        page = client.scroll(scroll_id = sid, scroll = str(wait_time) + 'm')
        end = time.time()

        ## update wait_time using a weighted running average. 
        wait_time =  ( (end - start + 10) + float(wait_time * p) ) / (p+1) 
        print "\t\t Page %s took %s seconds. We change the time to %s" %(p, end - start, wait_time)

        sid = page['_scroll_id'] # Update the scroll ID
        scroll_size = len(page["hits"]["hits"]) ## no. of hits returned on this page

        print "\t\t Page %s has returned %s hits. Storing .." %( p, scroll_size )
        returned_hits[p] = page['hits']['hits']

        doc_count += scroll_size ## update the total count of docs processed
        print "\t\t Returned and stored %s docs of %s \n" %(doc_count, total_hits)

    tries = 3   ## set tries to three so we exit the while loop! 

except: 
    e = sys.exc_info()[0]
    print "\t\t ---- Error on try %s\n\t\t size was %s, wait_time was %s min, \n\t\terror message = %s" %(tries , _size, wait_time, e) 

    tries += 1 ## increment tries, and do it again until 3 tries. 
    # wait_time *= 2 ## double the time interval for the next go round 
    size = int(.8 * size) ## lower size of docs per shard returned. 

    if tries == 3: 
        print "\t\t three strikes and you're out! (failed three times in a row to execute the alert query). Exiting. "

    else: 
        print '\t\t ---- trying again for the %s-th time ...' %( tries + 1 )