第一阶段 07 · 增删改与批量速查与示例（insert / update / delete / bulk）第一阶段

阶段第一阶段 / 核心概念目标把写操作新增、修改、删除、批量逐个用例子写出来和第 06 篇查询速查配套。本篇是独立文档示例不依赖任何具体项目。约定文档实体统一用OrderDoc定义见第 06 篇 §0.1。深入版单文档写入见第 30 篇、批量见第 31 篇、条件更新/删除见第 32 篇。0. 统一约定0.1 写操作执行器Slf4jComponentpublicclassOrderWriter{AutowiredprivateElasticsearchClientclient;privatestaticfinalStringINDEXorders_idx;// 下面各例的方法都放在这个类里}0.2 写 API 与 SQL 对照ES API语义SQL 对照index有则整篇覆盖、无则新建INSERT ... ON CONFLICT DO UPDATE整行替换create仅当不存在时创建存在报错INSERT主键冲突报错update局部更新指定字段 / 脚本更新UPDATE ... SET col ?delete按_id删除DELETE ... WHERE id ?updateByQuery按条件批量改UPDATE ... WHERE ...deleteByQuery按条件批量删DELETE ... WHERE ...bulk多条写打成一个请求批量INSERT/COPY1. 新增 / 覆盖index与create1.1index—— upsert 整篇最常用有则整篇覆盖、无则新建。显式指定_id保证幂等。/** 相当于 INSERT ... ON CONFLICT (id) DO UPDATE整行替换 */publicvoidindexDoc(OrderDocdoc)throwsIOException{client.index(i-i.index(INDEX).id(doc.getId())// 用业务主键做 _id.document(doc));// 直接传实体}1.2create—— 仅新建已存在则报错/** 相当于 INSERT主键冲突会抛异常 */publicvoidcreateDoc(OrderDocdoc)throwsIOException{client.create(c-c.index(INDEX).id(doc.getId()).document(doc));}index与create的区别index会覆盖已有文档create遇到相同_id直接报错。2. 修改update2.1 局部更新指定字段UPDATE ... SET只改传入的字段其它字段不动。用一个「部分对象」承载要改的字段。/** 相当于 UPDATE orders SET status ?, amount ? WHERE id ? */publicvoidupdateFields(Stringid,MapString,Objectpartial)throwsIOException{client.update(u-u.index(INDEX).id(id).doc(partial),// 只更新给出的字段OrderDoc.class);// 返回类型}// 调用只改 status 和 amountupdateFields(1001,Map.of(status,SHIPPED,amount,newBigDecimal(1299.00)));想用强类型也可以.doc(new OrderDoc())只 set 要改的字段——但对象里为 null 的字段也会被当成「不更新」还是「设为 null」取决于序列化配置稳妥起见局部更新用Map更可控。2.2 upsert不存在就插入存在就更新publicvoidupsert(OrderDocdoc,MapString,Objectpartial)throwsIOException{client.update(u-u.index(INDEX).id(doc.getId()).doc(partial)// 存在 → 局部更新这些字段.upsert(doc),// 不存在 → 用整个 doc 新建OrderDoc.class);}2.3 脚本更新原子自增等需要「基于旧值计算」时用脚本避免读-改-写的并发问题。/** 相当于 UPDATE orders SET amount amount 100 WHERE id ? */publicvoidincrAmount(Stringid,doubledelta)throwsIOException{client.update(u-u.index(INDEX).id(id).script(sc-sc.inline(in-in.source(ctx._source.amount params.d).params(d,JsonData.of(delta)))),OrderDoc.class);}3. 删除delete/** 相当于 DELETE FROM orders WHERE id ? */publicvoiddeleteDoc(Stringid)throwsIOException{client.delete(d-d.index(INDEX).id(id));}4. 条件批量改 / 删updateByQuery/deleteByQuery按条件一次处理一批不用先查出来再逐条改。4.1deleteByQuery—— 条件删除/** 相当于 DELETE FROM orders WHERE status CANCELLED */publicvoiddeleteCancelled()throwsIOException{client.deleteByQuery(d-d.index(INDEX).query(q-q.term(t-t.field(status).value(CANCELLED))));}4.2updateByQuery—— 条件更新脚本/** 相当于 UPDATE orders SET status ARCHIVED WHERE region AP */publicvoidarchiveApOrders()throwsIOException{client.updateByQuery(u-u.index(INDEX).query(q-q.term(t-t.field(region).value(AP))).script(sc-sc.inline(in-in.source(ctx._source.status ARCHIVED))));}5. 批量处理bulk把多条写操作打成一个请求发出去是高吞吐写入的标准做法。5.1 批量新增 / 覆盖/** 相当于批量 INSERT ... ON CONFLICT */publicvoidbulkIndex(ListOrderDocdocs)throwsIOException{ListBulkOperationopsnewArrayList(docs.size());for(OrderDocdoc:docs){ops.add(BulkOperation.of(b-b.index(i-i.index(INDEX).id(doc.getId()).document(doc))));}BulkResponserespclient.bulk(b-b.operations(ops));// 关键bulk 整体 200 不代表每条都成功必须逐条检查if(resp.errors()){resp.items().stream().filter(it-it.error()!null).forEach(it-log.error(bulk 失败 id{}, reason{},it.id(),it.error().reason()));}}5.2 一个 bulk 里混合 增 / 改 / 删publicvoidbulkMixed(ListOrderDoctoIndex,MapString,MapString,ObjecttoUpdate,ListStringtoDelete)throwsIOException{ListBulkOperationopsnewArrayList();// 新增/覆盖for(OrderDocdoc:toIndex){ops.add(BulkOperation.of(b-b.index(i-i.index(INDEX).id(doc.getId()).document(doc))));}// 局部更新toUpdate.forEach((id,partial)-ops.add(BulkOperation.of(b-b.update(u-u.index(INDEX).id(id).action(a-a.doc(partial))))));// 删除for(Stringid:toDelete){ops.add(BulkOperation.of(b-b.delete(d-d.index(INDEX).id(id))));}BulkResponserespclient.bulk(b-b.operations(ops));if(resp.errors()){resp.items().stream().filter(it-it.error()!null).forEach(it-log.error(bulk 失败 id{}, reason{},it.id(),it.error().reason()));}}相关 importco.elastic.clients.elasticsearch.core.bulk.BulkOperation、co.elastic.clients.elasticsearch.core.BulkResponse。6. 速查总表需求方法SQL 类比本篇深入新增/覆盖单条client.index(...)INSERT ... ON CONFLICT§1.1第 30 篇仅新建单条client.create(...)INSERT冲突报错§1.2第 30 篇改单条部分字段client.update(...).doc(...)UPDATE ... SET§2.1第 30 篇upsertclient.update(...).upsert(...)INSERT ... ON CONFLICT§2.2第 30 篇脚本更新client.update(...).script(...)SET x x ?§2.3第 30 篇删单条client.delete(...)DELETE WHERE id?§3第 30 篇条件删client.deleteByQuery(...)DELETE WHERE ...§4.1第 32 篇条件改client.updateByQuery(...)UPDATE WHERE ...§4.2第 32 篇批量client.bulk(...)批量INSERT/COPY§5第 31 篇7. 坑与最佳实践显式指定_id用业务主键做_id重跑不产生重复文档幂等。不指定会随机生成 id。index是整篇替换漏传字段会丢失只改部分字段用update。bulk 必须查resp.errors()整体 200 不代表每条成功要逐条检查items()。批大小适中一批几 MB / 几千条为宜太大易超时/OOM太小失去批量意义。写入近实时NRT默认 1s 后可搜到refresh_interval不是立即可见。大批导入临时调优可临时设refresh_interval: -1、number_of_replicas: 0完成后恢复。并发更新用乐观锁if_seq_no/if_primary_term防丢更新自增类改动优先用脚本更新。updateByQuery/deleteByQuery是重操作大范围执行会扫描很多文档注意配slices并行与错峰。第一阶段小结到这里第一阶段00–07完成你应该能用 SQL 心智理解 index / document / mapping01、02分清text/keyword理解分词与_score02、03独立配置并使用官方ElasticsearchClient04看懂查询相关类并写复杂bool动态拼接05熟练用各类查询06与各类写操作07结果/文档统一用实体类下一篇进入第二阶段查询能力10-match-全文匹配.md。写操作的深入讲解幂等、乐观锁、bulk 调优、条件更新见第四阶段 30–33 篇。