摘要：注意當(dāng)一個(gè)文檔在快照的時(shí)間和索引請(qǐng)求過程之間發(fā)生變化時(shí)，會(huì)發(fā)生版本沖突。當(dāng)版本匹配時(shí)，更新文檔并增加版本號(hào)。在正在運(yùn)行的更新中，使用更改的值使用查找的值。值加快進(jìn)程立即生效，減慢查詢的值在完成當(dāng)前批處理后生效，以防止?jié)L動(dòng)超時(shí)。

本節(jié)描述以下CRUD API:

Index API

Get API

Delete API

Update API

Multi Get API

Bulk API

Reindex API

Update By Query API

Delete By Query API

注意
所有CRUD API都是單索引API，索引參數(shù)接受單個(gè)索引名，或指向單個(gè)索引的別名

index API允許將類型化的JSON文檔索引到特定的索引中，并使其可搜索。

生成JSON文檔有幾種不同的方法:

手動(dòng)的（也就是你自己）使用原生byte[]或作為String

使用一個(gè)Map，該Map將自動(dòng)轉(zhuǎn)換為它的JSON等效項(xiàng)

使用第三方庫(kù)對(duì)bean(如Jackson)進(jìn)行序列化

使用內(nèi)置的助手XContentFactory.jsonBuilder()

在內(nèi)部，每個(gè)類型被轉(zhuǎn)換為byte[]（像String被轉(zhuǎn)換為byte[]），因此，如果對(duì)象已經(jīng)以這種形式存在，那么就使用它，jsonBuilder是高度優(yōu)化的JSON生成器，它直接構(gòu)造一個(gè)byte[]。

這里沒有什么困難，但是請(qǐng)注意，您必須根據(jù)日期格式對(duì)日期進(jìn)行編碼。

String json = "{" +
        ""user":"kimchy"," +
        ""postDate":"2013-01-30"," +
        ""message":"trying out Elasticsearch"" +
    "}";

Map是一個(gè)鍵:值對(duì)集合，它表示一個(gè)JSON結(jié)構(gòu):

Map json = new HashMap();
json.put("user","kimchy");
json.put("postDate",new Date());
json.put("message","trying out Elasticsearch");

可以使用Jackson將bean序列化為JSON，請(qǐng)將Jackson Databind添加到您的項(xiàng)目中，然后，您可以使用ObjectMapper來序列化您的bean:

import com.fasterxml.jackson.databind.*;

// instance a json mapper
ObjectMapper mapper = new ObjectMapper(); // create once, reuse

// generate json
byte[] json = mapper.writeValueAsBytes(yourbeaninstance);

Elasticsearch提供了內(nèi)置的助手來生成JSON內(nèi)容。

import static org.elasticsearch.common.xcontent.XContentFactory.*;

XContentBuilder builder = jsonBuilder()
    .startObject()
        .field("user", "kimchy")
        .field("postDate", new Date())
        .field("message", "trying out Elasticsearch")
    .endObject()

注意，您還可以使用startArray(String)和endArray()方法添加數(shù)組，順便說一下，field方法接受許多對(duì)象類型，您可以直接傳遞數(shù)字、日期甚至其他XContentBuilder對(duì)象。

如果需要查看生成的JSON內(nèi)容，可以使用string()方法。

String json = builder.string();

下面的示例將JSON文檔索引為一個(gè)名為twitter的索引，其類型為tweet, id值為1:

import static org.elasticsearch.common.xcontent.XContentFactory.*;

IndexResponse response = client.prepareIndex("twitter", "tweet", "1")
        .setSource(jsonBuilder()
                    .startObject()
                        .field("user", "kimchy")
                        .field("postDate", new Date())
                        .field("message", "trying out Elasticsearch")
                    .endObject()
                  )
        .get();

注意，您還可以將文檔索引為JSON字符串，并且不需要提供ID:

String json = "{" +
        ""user":"kimchy"," +
        ""postDate":"2013-01-30"," +
        ""message":"trying out Elasticsearch"" +
    "}";

IndexResponse response = client.prepareIndex("twitter", "tweet")
        .setSource(json, XContentType.JSON)
        .get();

IndexResponse對(duì)象會(huì)給你一個(gè)響應(yīng):

// Index name
String _index = response.getIndex();
// Type name
String _type = response.getType();
// Document ID (generated or not)
String _id = response.getId();
// Version (if it"s the first time you index this document, you will get: 1)
long _version = response.getVersion();
// status has stored current instance statement.
RestStatus status = response.status();

有關(guān)索引操作的更多信息，請(qǐng)查看REST索引文檔

get API允許根據(jù)索引的id從索引中獲取類型化的JSON文檔，下面的示例從一個(gè)名為twitter的索引中獲取JSON文檔，該索引的類型名為tweet, id值為1:

GetResponse response = client.prepareGet("twitter", "tweet", "1").get();

有關(guān)get操作的更多信息，請(qǐng)查看REST get文檔。

delete API允許基于id從特定索引中刪除類型化的JSON文檔，下面的示例從名為twitter的索引中刪除JSON文檔，該索引的類型名為tweet, id值為1:

DeleteResponse response = client.prepareDelete("twitter", "tweet", "1").get();

通過查詢刪除的API可以根據(jù)查詢結(jié)果刪除給定的一組文檔:

BulkByScrollResponse response = DeleteByQueryAction.INSTANCE.newRequestBuilder(client)
    .filter(QueryBuilders.matchQuery("gender", "male")) 
    .source("persons")                                  
    .get();                                             
long deleted = response.getDeleted();

QueryBuilders.matchQuery("gender", "male")（查詢）

source("persons") （索引）

get()（執(zhí)行操作）

response.getDeleted()（被刪除的文檔數(shù)）

由于這是一個(gè)長(zhǎng)時(shí)間運(yùn)行的操作，如果您希望異步執(zhí)行，可以調(diào)用execute而不是get，并提供如下監(jiān)聽器:

DeleteByQueryAction.INSTANCE.newRequestBuilder(client)
    .filter(QueryBuilders.matchQuery("gender", "male"))     
    .source("persons")                                      
    .execute(new ActionListener() {   
        @Override
        public void onResponse(BulkByScrollResponse response) {
            long deleted = response.getDeleted();           
        }
        @Override
        public void onFailure(Exception e) {
            // Handle the exception
        }
    });

您可以創(chuàng)建一個(gè)UpdateRequest并將其發(fā)送給客戶端:

UpdateRequest updateRequest = new UpdateRequest();
updateRequest.index("index");
updateRequest.type("type");
updateRequest.id("1");
updateRequest.doc(jsonBuilder()
        .startObject()
            .field("gender", "male")
        .endObject());
client.update(updateRequest).get();

也可以使用prepareUpdate()方法:

client.prepareUpdate("ttl", "doc", "1")
        .setScript(new Script("ctx._source.gender = "male""  , ScriptService.ScriptType.INLINE, null, null))
        .get();

client.prepareUpdate("ttl", "doc", "1")
        .setDoc(jsonBuilder()               
            .startObject()
                .field("gender", "male")
            .endObject())
        .get();

Script()（你的腳本，它也可以是本地存儲(chǔ)的腳本名）

setDoc()（將合并到現(xiàn)有的文檔）

注意，您不能同時(shí)提供腳本和doc

update API允許基于提供的腳本更新文檔:

UpdateRequest updateRequest = new UpdateRequest("ttl", "doc", "1")
        .script(new Script("ctx._source.gender = "male""));
client.update(updateRequest).get();

update API還支持傳遞一個(gè)部分文檔合并到現(xiàn)有文檔中（簡(jiǎn)單的遞歸合并，內(nèi)部合并對(duì)象，取代核心的“鍵/值”和數(shù)組），例如：

UpdateRequest updateRequest = new UpdateRequest("index", "type", "1")
        .doc(jsonBuilder()
            .startObject()
                .field("gender", "male")
            .endObject());
client.update(updateRequest).get();

也有對(duì)Upsert的支持，如果文檔不存在，則使用upsert元素的內(nèi)容索引新的doc:

IndexRequest indexRequest = new IndexRequest("index", "type", "1")
        .source(jsonBuilder()
            .startObject()
                .field("name", "Joe Smith")
                .field("gender", "male")
            .endObject());
UpdateRequest updateRequest = new UpdateRequest("index", "type", "1")
        .doc(jsonBuilder()
            .startObject()
                .field("gender", "male")
            .endObject())
        .upsert(indexRequest);              
client.update(updateRequest).get();

如果文檔不存在，將添加indexRequest中的文檔。

如果文件index/type/1已經(jīng)存在，我們將在此操作后獲得如下文件:

{
    "name"  : "Joe Dalton",
    "gender": "male"        
}

"gender": "male"（此字段由更新請(qǐng)求添加）

如果不存在，我們將有一份新文件:

{
    "name" : "Joe Smith",
    "gender": "male"
}

multi get API允許根據(jù)文檔的index、type和id獲取文檔列表:

MultiGetResponse multiGetItemResponses = client.prepareMultiGet()
    .add("twitter", "tweet", "1")           
    .add("twitter", "tweet", "2", "3", "4") 
    .add("another", "type", "foo")          
    .get();

for (MultiGetItemResponse itemResponse : multiGetItemResponses) { 
    GetResponse response = itemResponse.getResponse();
    if (response.isExists()) {                      
        String json = response.getSourceAsString(); 
    }
}

add("twitter", "tweet", "1")（通過單一id）

add("twitter", "tweet", "2", "3", "4")（或以相同index/type的id列表）

add("another", "type", "foo")（你也可以從另一個(gè)索引中得到）

MultiGetItemResponse itemResponse : multiGetItemResponses（迭代結(jié)果集）

response.isExists()（您可以檢查文檔是否存在）

response.getSourceAsString()（訪問_source字段）

有關(guān)multi get操作的更多信息，請(qǐng)查看剩余的multi get文檔

bulk API允許在一個(gè)請(qǐng)求中索引和刪除多個(gè)文檔，這里有一個(gè)示例用法:

import static org.elasticsearch.common.xcontent.XContentFactory.*;

BulkRequestBuilder bulkRequest = client.prepareBulk();

// either use client#prepare, or use Requests# to directly build index/delete requests
bulkRequest.add(client.prepareIndex("twitter", "tweet", "1")
        .setSource(jsonBuilder()
                    .startObject()
                        .field("user", "kimchy")
                        .field("postDate", new Date())
                        .field("message", "trying out Elasticsearch")
                    .endObject()
                  )
        );

bulkRequest.add(client.prepareIndex("twitter", "tweet", "2")
        .setSource(jsonBuilder()
                    .startObject()
                        .field("user", "kimchy")
                        .field("postDate", new Date())
                        .field("message", "another post")
                    .endObject()
                  )
        );

BulkResponse bulkResponse = bulkRequest.get();
if (bulkResponse.hasFailures()) {
    // process failures by iterating through each bulk response item
}

BulkProcessor類提供了一個(gè)簡(jiǎn)單的接口，可以根據(jù)請(qǐng)求的數(shù)量或大小自動(dòng)刷新bulk操作，或者在給定的時(shí)間之后。

要使用它，首先創(chuàng)建一個(gè)BulkProcessor實(shí)例:

import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;

BulkProcessor bulkProcessor = BulkProcessor.builder(
        client,  
        new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId,
                                   BulkRequest request) { ... } 

            @Override
            public void afterBulk(long executionId,
                                  BulkRequest request,
                                  BulkResponse response) { ... } 

            @Override
            public void afterBulk(long executionId,
                                  BulkRequest request,
                                  Throwable failure) { ... } 
        })
        .setBulkActions(10000) 
        .setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB)) 
        .setFlushInterval(TimeValue.timeValueSeconds(5)) 
        .setConcurrentRequests(1) 
        .setBackoffPolicy(
            BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3)) 
        .build();

beforeBulk()

此方法在執(zhí)行bulk之前被調(diào)用，例如，您可以通過request.numberOfActions()查看numberOfActions

afterBulk(...BulkResponse response)

此方法在執(zhí)行bulk之后被調(diào)用，例如，您可以通過response.hasFailures()檢查是否存在失敗請(qǐng)求

afterBulk(...Throwable failure)

當(dāng)bulk失敗并引發(fā)一個(gè)可拋出對(duì)象時(shí)，將調(diào)用此方法

setBulkActions(10000)

我們希望每10,000個(gè)請(qǐng)求就執(zhí)行一次bulk

setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB))

我們希望每5MB就flush一次

setFlushInterval(TimeValue.timeValueSeconds(5))

無(wú)論請(qǐng)求的數(shù)量是多少，我們都希望每5秒flush一次

setConcurrentRequests(1)

設(shè)置并發(fā)請(qǐng)求的數(shù)量，值為0意味著只允許執(zhí)行一個(gè)請(qǐng)求，在積累新的bulk請(qǐng)求時(shí)，允許執(zhí)行一個(gè)值為1的并發(fā)請(qǐng)求

setBackoffPolicy()

設(shè)置一個(gè)自定義的備份策略，該策略最初將等待100ms，以指數(shù)形式增加并重試三次，當(dāng)一個(gè)或多個(gè)bulk項(xiàng)目請(qǐng)求以EsRejectedExecutionException失敗時(shí)，將嘗試重試，該異常表明用于處理請(qǐng)求的計(jì)算資源太少，要禁用backoff，請(qǐng)傳遞BackoffPolicy.noBackoff()

默認(rèn)情況下，BulkProcessor:

bulkActions設(shè)置為1000

bulkSize設(shè)置為5mb

不設(shè)置flushInterval

將concurrentrequest設(shè)置為1，這意味著flush操作的異步執(zhí)行

將backoffPolicy設(shè)置為一個(gè)指數(shù)備份，8次重試，啟動(dòng)延時(shí)為50ms，總等待時(shí)間約為5.1秒

然后您可以簡(jiǎn)單地將您的請(qǐng)求添加到BulkProcessor:

bulkProcessor.add(new IndexRequest("twitter", "tweet", "1").source(/* your doc here */));
bulkProcessor.add(new DeleteRequest("twitter", "tweet", "2"));

當(dāng)所有的文檔都被加載到BulkProcessor，可以使用awaitClose或close方法進(jìn)行關(guān)閉:

bulkProcessor.awaitClose(10, TimeUnit.MINUTES);

或

bulkProcessor.close();

如果通過設(shè)置flushInterval來調(diào)度其他計(jì)劃的flush，這兩種方法都將flush所有剩余的文檔，并禁用所有其他計(jì)劃flush。如果并發(fā)請(qǐng)求被啟用，那么awaitClose方法等待指定的超時(shí)以完成所有bulk請(qǐng)求，然后返回true，如果在所有bulk請(qǐng)求完成之前指定的等待時(shí)間已經(jīng)過去，則返回false，close方法不等待任何剩余的批量請(qǐng)求完成并立即退出。

如果您正在使用Elasticsearch運(yùn)行測(cè)試，并且正在使用BulkProcessor來填充數(shù)據(jù)集，那么您最好將并發(fā)請(qǐng)求的數(shù)量設(shè)置為0，以便以同步方式執(zhí)行批量的flush操作:

BulkProcessor bulkProcessor = BulkProcessor.builder(client, new BulkProcessor.Listener() { /* Listener methods */ })
        .setBulkActions(10000)
        .setConcurrentRequests(0)
        .build();

// Add your requests
bulkProcessor.add(/* Your requests */);

// Flush any remaining requests
bulkProcessor.flush();

// Or close the bulkProcessor if you don"t need it anymore
bulkProcessor.close();

// Refresh your indices
client.admin().indices().prepareRefresh().get();

// Now you can start searching!
client.prepareSearch().get();

updateByQuery最簡(jiǎn)單的用法是在不更改源的情況下更新索引中的每個(gè)文檔，這種用法允許獲取一個(gè)新屬性或另一個(gè)在線映射更改。

UpdateByQueryRequestBuilder updateByQuery = UpdateByQueryAction.INSTANCE.newRequestBuilder(client);
updateByQuery.source("source_index").abortOnVersionConflict(false);
BulkByScrollResponse response = updateByQuery.get();

對(duì)updateByQuery API的調(diào)用從獲取索引快照開始，索引使用內(nèi)部版本控制找到任何文檔。

注意
當(dāng)一個(gè)文檔在快照的時(shí)間和索引請(qǐng)求過程之間發(fā)生變化時(shí)，會(huì)發(fā)生版本沖突。

當(dāng)版本匹配時(shí)，updateByQuery更新文檔并增加版本號(hào)。

所有更新和查詢失敗都會(huì)導(dǎo)致updateByQuery中止，這些故障可以從BulkByScrollResponse#getIndexingFailures方法中獲得，任何成功的更新仍然存在，并且不會(huì)回滾，當(dāng)?shù)谝淮问?dǎo)致中止時(shí)，響應(yīng)包含由失敗的bulk請(qǐng)求生成的所有失敗。

為了防止版本沖突導(dǎo)致updateByQuery中止，請(qǐng)?jiān)O(shè)置abortOnVersionConflict(false)，第一個(gè)示例之所以這樣做，是因?yàn)樗噲D獲取在線映射更改，而版本沖突意味著在相同時(shí)間開始updateByQuery和試圖更新文檔的沖突文檔。這很好，因?yàn)樵摳聦@取在線映射更新。

UpdateByQueryRequestBuilder API支持過濾更新的文檔，限制要更新的文檔總數(shù)，并使用腳本更新文檔：

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

UpdateByQueryRequestBuilder還允許直接訪問用于選擇文檔的查詢，您可以使用此訪問來更改默認(rèn)的滾動(dòng)大小，或者以其他方式修改對(duì)匹配文檔的請(qǐng)求。

UpdateByQueryRequestBuilder updateByQuery = UpdateByQueryAction.INSTANCE.newRequestBuilder(client);
updateByQuery.source("source_index")
    .source().setSize(500);
BulkByScrollResponse response = updateByQuery.get();

您還可以將大小與排序相結(jié)合以限制文檔的更新:

UpdateByQueryRequestBuilder updateByQuery = UpdateByQueryAction.INSTANCE.newRequestBuilder(client);
updateByQuery.source("source_index").size(100)
    .source().addSort("cat", SortOrder.DESC);
BulkByScrollResponse response = updateByQuery.get();

除了更改文檔的_source字段外，還可以使用腳本更改操作，類似于Update API:

UpdateByQueryRequestBuilder updateByQuery = UpdateByQueryAction.INSTANCE.newRequestBuilder(client);
updateByQuery.source("source_index")
    .script(new Script(
        ScriptType.INLINE,
        "if (ctx._source.awesome == "absolutely) {"
            + "  ctx.op="noop""
            + "} else if (ctx._source.awesome == "lame") {"
            + "  ctx.op="delete""
            + "} else {"
            + "ctx._source.awesome = "absolutely"}",
        "painless",
        Collections.emptyMap()));
BulkByScrollResponse response = updateByQuery.get();

在Update API中，可以設(shè)置ctx.op的值來更改執(zhí)行的操作:

noop

如果您的腳本沒有做任何更改，設(shè)置ctx.op = "noop"，updateByQuery操作將從更新中省略該文檔，這種行為增加了響應(yīng)主體中的noop計(jì)數(shù)器。

delete

如果您的腳本決定必須刪除該文檔，設(shè)置ctx.op = "delete"，刪除將在響應(yīng)主體中已刪除的計(jì)數(shù)器中報(bào)告。

將ctx.op設(shè)置為任何其他值都會(huì)產(chǎn)生錯(cuò)誤，在ctx中設(shè)置任何其他字段都會(huì)產(chǎn)生錯(cuò)誤。

這個(gè)API不允許您移動(dòng)它所接觸的文檔，只是修改它們的源，這是故意的!我們沒有規(guī)定要把文件從原來的位置移走。

您也可以同時(shí)對(duì)多個(gè)索引和類型執(zhí)行這些操作，類似于search API:

UpdateByQueryRequestBuilder updateByQuery = UpdateByQueryAction.INSTANCE.newRequestBuilder(client);
updateByQuery.source("foo", "bar").source().setTypes("a", "b");
BulkByScrollResponse response = updateByQuery.get();

如果提供路由值，則進(jìn)程將路由值復(fù)制到滾動(dòng)查詢，將進(jìn)程限制為與路由值匹配的碎片：

UpdateByQueryRequestBuilder updateByQuery = UpdateByQueryAction.INSTANCE.newRequestBuilder(client);
updateByQuery.source().setRouting("cat");
BulkByScrollResponse response = updateByQuery.get();

updateByQuery也可以通過指定這樣的pipeline來使用ingest節(jié)點(diǎn):

UpdateByQueryRequestBuilder updateByQuery = UpdateByQueryAction.INSTANCE.newRequestBuilder(client);
updateByQuery.setPipeline("hurray");
BulkByScrollResponse response = updateByQuery.get();

您可以使用Task API獲取所有正在運(yùn)行的update-by-query請(qǐng)求的狀態(tài):

ListTasksResponse tasksList = client.admin().cluster().prepareListTasks()
    .setActions(UpdateByQueryAction.NAME).setDetailed(true).get();
for (TaskInfo info: tasksList.getTasks()) {
    TaskId taskId = info.getTaskId();
    BulkByScrollTask.Status status = (BulkByScrollTask.Status) info.getStatus();
    // do stuff
}

使用上面所示的TaskId，您可以直接查找任務(wù):

GetTaskResponse get = client.admin().cluster().prepareGetTask(taskId).get();

使用Cancel Task API

任何查詢更新都可以使用Task Cancel API取消:

// Cancel all update-by-query requests
client.admin().cluster().prepareCancelTasks().setActions(UpdateByQueryAction.NAME).get().getTasks();
// Cancel a specific update-by-query request
client.admin().cluster().prepareCancelTasks().setTaskId(taskId).get().getTasks();

使用list tasks API查找taskId的值。

取消請(qǐng)求通常是一個(gè)非常快速的過程，但可能要花費(fèi)幾秒鐘的時(shí)間，task status API繼續(xù)列出任務(wù)，直到取消完成。

在正在運(yùn)行的更新中，使用_rethrottle API更改requests_per_second的值:

RethrottleAction.INSTANCE.newRequestBuilder(client)
    .setTaskId(taskId)
    .setRequestsPerSecond(2.0f)
    .get();

使用list tasks API查找taskId的值。

與updateByQuery API一樣，requests_per_second的值可以是任何正值的浮點(diǎn)值來設(shè)置節(jié)流的級(jí)別，或者Float.POSITIVE_INFINITY禁用節(jié)流。requests_per_second值加快進(jìn)程立即生效，減慢查詢的requests_per_second值在完成當(dāng)前批處理后生效，以防止?jié)L動(dòng)超時(shí)。

詳情見reindex API

BulkByScrollResponse response = ReindexAction.INSTANCE.newRequestBuilder(client)
    .destination("target_index")
    .filter(QueryBuilders.matchQuery("category", "xzy")) 
    .get();

還可以提供查詢來篩選應(yīng)該從源索引到目標(biāo)索引的哪些文檔。