Elasticsearch批量API

Question

我想问一下关于Elasticsearch批量API的问题

这是我使用批量API的代码

public void bulkInsert(String index, ArrayList<String> jsonList) throws IOException {
    BulkRequest request = new BulkRequest(); 

    for(String json: jsonList){
        if(json != null&& !json.isEmpty()){
            request.add(new IndexRequest(index)  
                    .source(json, XContentType.JSON));  
        }
    }

    BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);
    for (BulkItemResponse bulkItemResponse : bulkResponse) { 
        DocWriteResponse itemResponse = bulkItemResponse.getResponse(); 

        switch (bulkItemResponse.getOpType()) {
        case INDEX:    
        case CREATE:
            IndexResponse indexResponse = (IndexResponse) itemResponse;
            break;
        case UPDATE:   
            UpdateResponse updateResponse = (UpdateResponse) itemResponse;
            break;
        case DELETE:   
            DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
        }
    }
    if (bulkResponse.hasFailures()) { 
        for (BulkItemResponse bulkItemResponse : bulkResponse) {
            if (bulkItemResponse.isFailed()) { 
                BulkItemResponse.Failure failure =
                        bulkItemResponse.getFailure(); 

                System.out.println("failed: " + failure.getId());

            }
        }
    }
}

我遇到了超时异常，因为我的记录有800K。java.net.SocketTimeoutException：连接超时30,000毫秒http-outgoing-16[活动]

我试图分解传入的jsonList，但有时会出现相同的错误。

我目前使用的是Elasticsearch 7.6.2版本。

异常跟踪

Java.net.SocketTimeoutException：连接http-outgoing-16[ACTIVE]，位于org.elasticsearch.client.RestClient.ExtractandWrapcause（RestClient.Java:808)处org.elasticsearch.client.RestClient.PerformRequest（RestClient.Java:248)处org.elasticsearch.client.RestClient.PerformRequest（RestClient.Java:235)处org.elasticsearch.client.RestHighLevelClient.InternalPerformRequesta.lang.Reflect.Method.Invoke（未知源）在org.eclipse.jdt.internal.jarinjarloader.jarrsrcloader.main（jarrsrcloader.java:58)处，原因是：java.net.sockettimeoutexception：30,000毫秒超时连接http-outgoing-16[ACTIVE]在org.apache.http.nio.protocol.httpasyncrequestexecutor.timeout(httpasyncrequestexecutor.387)在nio.reactor.baseioreactor.execute（baseioreactor.java:104)在org.apache.http.impl.nio.reactor.abstractmultiworkerioreactor$worker.run（abstractmultiworkerioreactor.java:591)在java.lang.thread.run（未知源）

Answer 1

由于您正在使用批量API并向Elasticsearch发送大量数据，并且连接的默认超时为30秒，而Elasticsearch无法在30秒内完成这个巨大的批量操作，因此会出现此异常。

这种超时对于大容量API是正常的，在您的情况下（特定于索引），可以执行以下操作：

扩展集群即添加更多的CPU、内存、更好的磁盘、禁用refresh_interval（默认值为1秒）以加快批量索引。

如正式ES文件所述

request.timeout(TimeValue.timeValueMinutes(2));   --> 2 min timeout
request.timeout("2m");  --> string format of 2 sec.

编辑：如评论中所问，您可以使用批量API的同步执行，如果您想立即检查您的批量API的响应，下面引用自同一文档：

检索操作的响应（成功与否），可以是IndexResponse、UpdateResponse或DeleteResponse，都可以看作是DocWriteResponse实例