32 · update_by_query / delete_by_query 条件更新删除阶段第四阶段 / 写入与索引管理ESupdate_by_query/delete_by_query| PostgreSQLUPDATE ... WHERE/DELETE ... WHERE1. 概念前一篇的update/delete是按_id操作单文档。本篇的update_by_query/delete_by_query是按查询条件批量操作真正对应 SQL 的UPDATE ... WHERE和DELETE ... WHERE。更新逻辑通常用脚本painless表达。2. PostgreSQL 对照-- 条件更新UPDATEsalesdataSETstatusCLOSEDWHEREgeo_cdAPANDnet_amount0;-- 条件删除DELETEFROMsalesdataWHEREversion_numberv-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 statusCLOSED WHERE geo_cd? AND net_amount0 */publiclongcloseZeroAmount(StringindexName,StringgeoCd)throwsIOException{UpdateByQueryResponserespelasticsearchClient.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{DeleteByQueryResponserespelasticsearchClient.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. 坑与最佳实践是异步分批的批量操作数据量大时耗时长可用wait_for_completionfalse拿 task id 异步跟踪。版本冲突默认遇冲突中断设conflicts: proceed可跳过冲突继续。脚本性能update_by_query逐文档跑脚本海量文档慢能提前算好用 bulk 更快。不可回滚删除/更新不可逆生产前先用相同query跑_count确认影响范围。refresh需要立即可见可带refreshtrue但会影响性能。