ElasticSearch update_by_query

àì夳堔傛蜴生んèń 2022-03-16 18:50 234阅读 0赞

项目中需要定时对ElasticSearch的某些数据进行更新，数据量十几万，采用RestClient更新，在没有设置超时的情况下，报错如下：java.io.IOException: listener timeout after waiting for \[30000\] ms

一般通常的做法都是修改超时时间设置：

restClient = RestClient.builder(
            new HttpHost(GSConstants.P_ES_HOST, GSConstants.P_ES_PORT, "http"))
            .setMaxRetryTimeoutMillis(TIMEOUT)
            .setHttpClientConfigCallback(new HttpClientConfigCallback(){
    			@Override
    			public HttpAsyncClientBuilder customizeHttpClient(
    				HttpAsyncClientBuilder httpClientBuilder) {
    				RequestConfig.Builder requestConfigBuilder = RequestConfig.custom()
    				               .setConnectTimeout(5*60*1000)//超时时间5分钟
    				               .setSocketTimeout(5*60*1000)//这就是Socket超时时间设置
    				               .setConnectionRequestTimeout(DEFAULT_CONNECTION_REQUEST_TIMEOUT_MILLIS);
    				httpClientBuilder.setDefaultRequestConfig(requestConfigBuilder.build());
    				return httpClientBuilder; 
    			}
    }).build();

[update\_by\_query ][update_by_query]顾名思义，通过查询更新，既先query出符合条件的内容再进行update操作，新增字段，修改字段值都可以满足，这个批量修改更符合我们的需求。

updateByQuery的调用首先获取索引的快照，对使用内部版本控制的任何文档进行索引。当版本匹配时，updateByQuery会更新文档并增加版本号。当一个文档在快照的时间和索引请求过程的时间之间发生变化时，就会发生版本冲突。为了防止版本冲突导致updateByQuery中止，设置终止冲突（false）。

所有更新和查询失败都会导致updateByQuery中止，任何成功的更新都是保留的，不会回滚。当第一个故障导致中止时，响应包含失败的批量请求所产生的所有故障。

UpdateByQueryRequestBuilder updateByQuery = UpdateByQueryAction.INSTANCE.newRequestBuilder(client);
    updateByQuery.source("source_index")
        .filter(QueryBuilders.termQuery("name", "alex"))
        .size(1000)
        .script(new Script(ScriptType.INLINE, "ctx._source.awesome = 'absolutely'", "painless", Collections.emptyMap()));
    BulkByScrollResponse response = updateByQuery.get();

您可以使用这个访问来改变默认的滚动大小，或者对匹配文档的请求进行修改。![85332335][]

[update_by_query]: https://www.elastic.co/guide/en/elasticsearch/reference/6.2/docs-update-by-query.html
[85332335]: /images/20220310/c3504d9698c145aab9b8bdaed10767f5.png