第四阶段 32 · update_by_query / delete_by_query 条件更新删除

32 · update_by_query / delete_by_query 条件更新删除

阶段：第四阶段 / 写入与索引管理
ES：update_by_query/delete_by_query| PostgreSQL：UPDATE ... WHERE/DELETE ... WHERE

1. 概念

前一篇的update/delete是按_id操作单文档。
本篇的update_by_query/delete_by_query是按查询条件批量操作，
真正对应 SQL 的UPDATE ... WHERE和DELETE ... WHERE。
更新逻辑通常用脚本（painless）表达。

2. PostgreSQL 对照

-- 条件更新UPDATEsalesdataSETstatus='CLOSED'WHEREgeo_cd='AP'ANDnet_amount=0;-- 条件删除DELETEFROMsalesdataWHEREversion_number='v-old';

3. ES DSL

# 条件更新：给匹配文档加字段 / 改值 POST salesdata_idx/_update_by_query { "query": { "bool": { "must": [ { "term": { "geo_cd": "AP" } }, { "term": { "net_amount": 0 } } ] } }, "script": { "source": "ctx._source.status = params.s", "params": { "s": "CLOSED" } } } # 条件删除 POST salesdata_idx/_delete_by_query { "query": { "term": { "version_number": "v-old" } } }

4. Spring Boot 实现

@ComponentpublicclassDoc32ByQuery{@AutowiredprivateElasticsearchClientelasticsearchClient;/** 条件更新：UPDATE ... SET status='CLOSED' WHERE geo_cd=? AND net_amount=0 */publiclongcloseZeroAmount(StringindexName,StringgeoCd)throwsIOException{UpdateByQueryResponseresp=elasticsearchClient.updateByQuery(u->u.index(indexName).query(q->q.bool(b->b.must(m->m.term(t->t.field("geo_cd").value(geoCd))).must(m->m.term(t->t.field("net_amount").value(0))))).script(sc->sc.inline(in->in.source("ctx._source.status = params.s").params("s",JsonData.of("CLOSED")))).conflicts(Conflicts.Proceed));// 冲突继续而非中断returnresp.updated();}/** 条件删除：DELETE WHERE version_number=? */publiclongdeleteByVersion(StringindexName,Stringversion)throwsIOException{DeleteByQueryResponseresp=elasticsearchClient.deleteByQuery(d->d.index(indexName).query(q->q.term(t->t.field("version_number").value(version))));returnresp.deleted();}}

import：...core.UpdateByQueryResponse、...core.DeleteByQueryResponse、
co.elastic.clients.json.JsonData、...core.search.Conflicts（或_types.Conflicts）。

5. 坑与最佳实践

1. 是异步分批的批量操作：数据量大时耗时长，可用wait_for_completion=false拿 task id 异步跟踪。
2. 版本冲突：默认遇冲突中断，设conflicts: proceed可跳过冲突继续。
3. 脚本性能：update_by_query逐文档跑脚本，海量文档慢；能提前算好用 bulk 更快。
4. 不可回滚：删除/更新不可逆，生产前先用相同query跑_count确认影响范围。
5. refresh：需要立即可见可带refresh=true，但会影响性能。