本文共 3390 字，大约阅读时间需要 11 分钟。

Elasticsearch Bulk API??

Bulk API?Elasticsearch?????????????????????API???????????????????????????????????????????????????????

BulkRequest??

BulkRequest ???Bulk API???????????????????????

????

????

?? add ??? BulkRequest ???????????????????

Bulk API????

Bulk API??????? JSON??????????

????

????

{ "opType" : { ... } }
{ "meta" : { ... } }
{ "payload" : { ... } }

?????

??????????????

• _index?????
• _type?????
• _id???ID
• routing????
• parent????ID
• version??????
• version_type?????

???????

??????

Bulk API????

????

????????????????????????????

??

???????????????????

??????

???? BulkRequest#waitForActiveShards ??????????????????????????

Bulk API??

????? Bulk API ?????????????

public static final void testBulk() {
    RestHighLevelClient client = EsClient.getClient();
    try {
        IndexRequest indexRequest = new IndexRequest("twitter", "_doc", "12")
                .source(buildTwitter("dingw", "2009-11-18T14:12:12", "test bulk"));
        
        UpdateRequest updateRequest = new UpdateRequest("twitter", "_doc", "11")
                .doc(new IndexRequest("twitter", "_doc", "11"))
                .source(buildTwitter("dingw", "2009-11-18T14:12:12", "test bulk update"));
        
        BulkRequest request = new BulkRequest();
        request.add(indexRequest);
        request.add(updateRequest);
        
        BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);
        
        for (BulkItemResponse bulkItemResponse : bulkResponse) {
            if (bulkItemResponse.isFailed()) {
                BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                System.out.println(failure);
                continue;
            }
            DocWriteResponse itemResponse = bulkItemResponse.getResponse();
            if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.INDEX ||
                bulkItemResponse.getOpType() == DocWriteRequest.OpType.CREATE) {
                IndexResponse indexResponse = (IndexResponse) itemResponse;
                System.out.println(indexRequest);
            } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.UPDATE) {
                UpdateResponse updateResponse = (UpdateResponse) itemResponse;
                System.out.println(updateRequest);
            } else if (bulkItemResponse.getOpType() == DocWriteRequest.OpType.DELETE) {
                DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
                System.out.println(deleteResponse);
            }
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        EsClient.close(client);
    }
}

??

???????????????? Bulk API ????????????Bulk API ? Elasticsearch ?????????????????????????????????????????