java - ElasticSearch 使用 Java RestHighLevelClient 通过 BulkRequest API 索引 100K 文档

Question

正在使用滚动 API 从索引 documents_qa 读取 100k 加上文件路径。实际文件将在我的本地 d:\drive 中提供。通过使用文件路径，我读取实际文件并转换为 Base64，并使用另一个索引 document_attachment_qa 中的(文件的)base64 内容重新索引。

我当前的实现是，读取filePath，将文件转换为base64，然后将文档与文件内容一一索引。因此，它需要更多时间，例如:索引 4000 个文档需要超过 6 个小时，而且连接由于 IO 异常而终止。

所以现在我想使用 BulkRequest API 索引文档，但我使用 RestHighLevelClient，并且不确定如何将 BulkRequest API 与 RestHighLevelClient 一起使用。

请找到我当前的实现，它正在逐个文档建立索引。

jsonMap = new HashMap<String, Object>();
            jsonMap.put("id", doc.getId());
            jsonMap.put("app_language", doc.getApp_language());
            jsonMap.put("fileContent", result);

            String id=Long.toString(doc.getId());

IndexRequest request = new IndexRequest(ATTACHMENT, "doc", id ) // ATTACHMENT is the index name
                    .source(jsonMap) // Its my single document.
                    .setPipeline(ATTACHMENT);

IndexResponse response = SearchEngineClient.getInstance3().index(request); // increased timeout 

我找到了以下 BulkRequest 文档。

https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/java-docs-bulk.html

但我不确定如何在使用 RestHighLevelClient 时实现 BulkRequestBuilderbulkRequest = client.prepareBulk(); client.prepareBulk() 方法。

更新1

我正在尝试一次性索引所有 100K 文档。所以我创建了一个 JSONArray 并将所有 JSONObject 一一放入数组中。最后，我尝试构建 BulkRequest 并将我的所有文档 (JSONArray) 添加为 BulkRequest 的源，并尝试对它们建立索引。

这里我不确定如何将我的 JSONArray 转换为字符串列表。

private final static String ATTACHMENT = "document_attachment_qa";
private final static String TYPE = "doc";
JSONArray reqJSONArray=new JSONArray();

while (searchHits != null && searchHits.length > 0) { 
...
...
    jsonMap = new HashMap<String, Object>();
    jsonMap.put("id", doc.getId());
    jsonMap.put("app_language", doc.getApp_language());
    jsonMap.put("fileContent", result);

    reqJSONArray.put(jsonMap)
}

String actionMetaData = String.format("{ \"index\" : { \"_index\" : \"%s\", \"_type\" : \"%s\" } }%n", ATTACHMENT, TYPE);
List<String> bulkData =   // not sure how to convert a list of my documents in JSON strings    
StringBuilder bulkRequestBody = new StringBuilder();
for (String bulkItem : bulkData) {
    bulkRequestBody.append(actionMetaData);
    bulkRequestBody.append(bulkItem);
    bulkRequestBody.append("\n");
}

HttpEntity entity = new NStringEntity(bulkRequestBody.toString(), ContentType.APPLICATION_JSON);
try {
    Response response = SearchEngineClient.getRestClientInstance().performRequest("POST", "/ATTACHMENT/TYPE/_bulk", Collections.emptyMap(), entity);
    return response.getStatusLine().getStatusCode() == HttpStatus.SC_OK;
} catch (Exception e) {
    // do something
}

Answer 1

除了@ Chengpohi 的回答。我想补充以下几点:

BulkRequest 可用于使用单个请求执行多个索引、更新和/或删除操作。

它至少需要将一项操作添加到批量请求中:

BulkRequest request = new BulkRequest(); 
request.add(new IndexRequest("posts", "doc", "1")  
        .source(XContentType.JSON,"field", "foo"));
request.add(new IndexRequest("posts", "doc", "2")  
        .source(XContentType.JSON,"field", "bar"));
request.add(new IndexRequest("posts", "doc", "3")  
        .source(XContentType.JSON,"field", "baz"));

Note: The Bulk API supports only documents encoded in JSON or SMILE. Providing documents in any other format will result in an error.

同步操作:

BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);

客户端将是高级休息客户端并且执行将是同步的。

异步操作(推荐方法):

client.bulkAsync(request, RequestOptions.DEFAULT, listener);

批量请求的异步执行需要将 BulkRequest 实例和 ActionListener 实例传递给异步方法。

Listener Example:

ActionListener<BulkResponse> listener = new ActionListener<BulkResponse>() {
    @Override
    public void onResponse(BulkResponse bulkResponse) {

    }

    @Override
    public void onFailure(Exception e) {

    }
};

返回的 BulkResponse 包含有关已执行操作的信息，并允许迭代每个结果，如下所示:

for (BulkItemResponse bulkItemResponse : bulkResponse) { 
    DocWriteResponse itemResponse = bulkItemResponse.getResponse(); 

    if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX
            || bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) { 
        IndexResponse indexResponse = (IndexResponse) itemResponse;

    } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) { 
        UpdateResponse updateResponse = (UpdateResponse) itemResponse;

    } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) { 
        DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
    }
}

可以选择提供以下参数:

request.timeout(TimeValue.timeValueMinutes(2)); 
request.timeout("2m");

我希望这会有所帮助。