ElasticSearch Update API

Posted 2020-10-19 一颗小蚕豆 期待发芽

一、修改部分字段By  UpdateRequest

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();

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();

 

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

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

二、乐观锁方式 指定Version 版本更新(es 5)

RefreshPolicy.IMMEDIATE 更新策略为立即更新，ElasticSearch 实际上是伪实时的，所有分片之间默认1s同步更新。

IndexResponse response = client.prepareIndex().setRefreshPolicy(RefreshPolicy.IMMEDIATE).setIndex(indexName)
.setType(tableName).setId(statid).setVersion(version).setSource(jsondata).execute().actionGet();

 

三、批量更新(es 5)

/**
* 批量索引
*
* @param indexName
* @param tableName
* @param maps
*/
public void bulkIndex(String indexName, String tableName, Map<String, String> maps) {

if (maps == null || maps.isEmpty()) {
return;
}

logger.info("[[title=bulkIndex]] start");
BulkRequestBuilder bulkRequest = client.prepareBulk().setRefreshPolicy(RefreshPolicy.IMMEDIATE);

int i = 1;
for (Map.Entry<String, String> entry : maps.entrySet()) {

bulkRequest.add(client.prepareIndex().setIndex(indexName).setType(tableName).setId(entry.getKey())
.setSource(entry.getValue()));

if (i % 100 == 0) {
bulkRequest.execute().actionGet();
}
i++;
}

bulkRequest.execute().actionGet();

String fieldString = SensitiveInfoMask.maskJsonSensitiveField(JSON.toJSONString(maps),
"(contactMobile|clientName|contactName|idCardNo)");
logger.info("[[title=bulkIndex]]批量索引信息={}", fieldString);
logger.info("[[title=bulkIndex]] end");
}

 

Upsertedit
There is also support for upsert. If the document does not exist, the content of the upsert element will be used to index the fresh 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();

If the document does not exist, the one in indexRequest will be added

If the document index/type/1 already exists, we will have after this operation a document like:

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

This field is added by the update request

If it does not exist, we will have a new document:

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