ElasticSearch批量更新问题：如何高效移除所有文档ingredients列表指定元素

高效解决方案：Elasticsearch 服务端批量更新（Update By Query + Painless 脚本） #

该方案完全在ES集群侧执行，不需要将全量文档拉取到业务端修改回写，同时会自动过滤不需要修改的文档，实际执行效率远高于客户端遍历方案。

原生DSL实现 #

POST /{替换为你的菜品索引名}/_update_by_query?conflicts=proceed&scroll_size=1000
{
  "query": {
    "term": {
      "ingredients": "{替换为要移除的食材名称}"
    }
  },
  "script": {
    "source": "ctx._source.ingredients.removeIf(ingredient -> ingredient == params.to_remove)",
    "lang": "painless",
    "params": {
      "to_remove": "{替换为要移除的食材名称}"
    }
  }
}

方案优势

• 仅处理ingredients字段包含待移除食材的文档，不会操作无关联数据，进一步降低实际处理量
• 所有计算、存储操作都在ES节点本地完成，没有跨网络传输的额外开销
• 自带批量、冲突处理能力，不需要额外开发逻辑

可选优化参数

• 数据量较小时可以去掉URL后的参数直接执行
• 数据量极大时可以加slices=auto参数开启并行执行，提升处理速度
• 生产高峰时段执行可以加throttle=1000参数限流（单位为次/秒），避免占用过多集群资源影响线上业务
• 若使用ES 7.0以下版本，不支持removeIf方法，可以将脚本替换为：

  "source": "ctx._source.ingredients = ctx._source.ingredients.stream().filter(i -> i != params.to_remove).collect(Collectors.toList())"
  

Spring Data Elasticsearch 实现（适配你提供的实体类） #

import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import static org.elasticsearch.index.query.QueryBuilders.termQuery;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.springframework.data.elasticsearch.core.UpdateByQueryRequestBuilder;

// 要移除的目标食材
String toRemoveIngredient = "xxx";

// 构造查询条件
NativeSearchQuery query = new NativeSearchQueryBuilder()
        .withQuery(termQuery("ingredients", toRemoveIngredient))
        .build();

// 构造执行脚本
Script script = new Script(
        ScriptType.INLINE,
        "painless",
        "ctx._source.ingredients.removeIf(ingredient -> ingredient == params.to_remove)",
        Collections.singletonMap("to_remove", toRemoveIngredient)
);

// 执行批量更新
UpdateByQueryRequest request = UpdateByQueryRequestBuilder.forQuery(query)
        .withScript(script)
        .withConflictsProceed(true)
        .withScrollSize(1000)
        .build();
elasticsearchOperations.updateByQuery(request, SOME_CLASS_NAME.class);

注意事项 #

• 执行前建议先小范围测试脚本逻辑，确认修改符合预期后再全量执行
• 执行结束后可以查看返回结果中的version_conflicts字段，确认跳过的冲突文档数量，按需单独处理即可

内容的提问来源于stack exchange，提问作者MrGa