ElasticSearch使用RestHighLevelClient批量插入gies错误

Question

使用RestHighLevelRestClient使用ElasticSearch批量插入时出现异常。

ElasticsearchStatusException[无法解析响应体]；html" target="_blank">嵌套：ResponseException[method[POST]，host[http:x.com]，URI[/_bulk？timeout=1m]，状态行[http/1.1 413请求实体太大]{“消息”：“请求大小超过104857600字节”}；在org。弹性搜索。客户RestHighLevelClient。org上的parseResponseException（RestHighLevelClient.java:1386）。弹性搜索。客户RestHighLevelClient 1美元。org上的onFailure（RestHighLevelClient.java:1357）。弹性搜索。客户RestClient$FailureTrackingResponseListener。org上的onDefinitiveFailure（RestClient.java:844）。弹性搜索。客户1美元。已在org上完成（RestClient.java:548）。弹性搜索。客户1美元。已在org上完成（RestClient.java:529）。阿帕奇。http。同时发生的基本未来。已在org上完成（BasicFuture.java:122）。阿帕奇。http。impl。尼奥。客户DefaultClientExchangeHandlerImpl。responseCompleted（DefaultClientExchangeHandlerImpl.java:181）位于org。阿帕奇。http。尼奥。协议HttpAsyncRequestExecutor。processResponse（HttpAsyncRequestExecutor.java:448）位于org。阿帕奇。http。尼奥。协议HttpAsyncRequestExecutor。org上的inputReady（HttpAsyncRequestExecutor.java:338）。阿帕奇。http。impl。尼奥。DefaultNHttpClientConnection。consumeInput（DefaultNHttpClientConnection.java:265）位于org。阿帕奇。http。impl。尼奥。客户斯帕奇。onInputReady（internaliodiispatch.java:81）位于org。阿帕奇。http。impl。尼奥。客户斯帕奇。onInputReady（internaliodiispatch.java:39）位于org。阿帕奇。http。impl。尼奥。核Reactor我是斯帕奇。inputReady（abstractioidispatch.java:114）位于org。阿帕奇。http。impl。尼奥。核Reactor基本反应器。可在org上阅读（BaseIOReactor.java:162）。阿帕奇。http。impl。尼奥。核Reactor抽象反应器。processEvent（AbstractIOReactor.java:337）位于org。阿帕奇。http。impl。尼奥。核Reactor抽象反应器。org上的processEvents（AbstractIOReactor.java:315）。阿帕奇。http。impl。尼奥。核Reactor抽象反应器。在org上执行（AbstractIOReactor.java:276）。阿帕奇。http。impl。尼奥。核Reactor基本反应器。在org上执行（BaseIOReactor.java:104）。阿帕奇。http。impl。尼奥。核ReactorAbstractMultiWorker$Worker。在java上运行（AbstractMultiworkerIOReactor.java:591）。朗。丝线。运行（未知源）被抑制：ParsingException[未能分析对象：应为名为[error]的字段，但在组织中找到[Message]]。弹性搜索。常见的非常重要。XContentParserUtils。ensureFieldName（XContentParserUtils.java:50）位于org。弹性搜索。弹性搜索例外。org上的failureFromXContent（ElasticsearchException.java:605）。弹性搜索。RestBytesResponse。org上的errorFromXContent（BytesRestResponse.java:169）。弹性搜索。客户RestHighLevelClient。org上的parseEntity（RestHighLevelClient.java:1406）。弹性搜索。客户RestHighLevelClient。parseResponseException（RestHighLevelClient.java:1382）。。。还有19个

我在SO post上搜索了一下，发现有些人建议设置http。最大内容长度，但我无法解决这个问题。下面是我保存数据的代码-

private void saveAll(BulkRequest bulkRequest, String indexName)
    {
        try {
            System.out.println("***Saving data into " + indexName + " of Size = " + bulkRequest.numberOfActions());
            // restHighLevelClient.bulk(bulkRequest,RequestOptions.DEFAULT);

             Builder builder = RequestOptions.DEFAULT.toBuilder();
             builder.addHeader("http.max_content_length", "500mb");

             RequestOptions requestOptions = builder.build();

            restHighLevelClient.bulkAsync(bulkRequest,requestOptions , new ActionListener<BulkResponse>()
            {

                @Override
                public void onResponse(BulkResponse response)
                {
                    System.out.println("Bulk Data sucessfully Saved in " + indexName + " ElasticIndex***");

                }

                @Override
                public void onFailure(Exception e)
                {
                    System.out.println("Issue in bulk data saving in " + indexName + " ElasticIndex***");
                    e.printStackTrace();

                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

Answer 1

我已经通过减少请求的大小解决了这个问题。您可以在下面的代码中看到，如果dataModelList size id大于10000，那么我将该列表拆分为5000个数据，然后保存到Elastic中。希望它能帮助别人。

 if (dataModelList.size() > 10000) {
            List<List<BaseDataModel>> baseDataModels = Lists.partition(dataModelList, 5000);
            baseDataModels.stream().forEach(baseModels -> {
                BulkRequest bulkRequest = new BulkRequest();
                baseModels.stream().forEach(baseModel -> {
                    bulkRequest.add(getBaseDataModelIndexRequest(baseModel));
                });
                saveAll(bulkRequest, rawEventIndex);
            });
        }