Spark写入Elasticsearch报错：无法写入全部条目，疑ES过载

这种Could not write all entries [13/128] (Maybe ES was overloaded?)的报错我之前帮不少开发者排查过，核心原因大概率是Elasticsearch扛不住你当前的写入压力，或者Spark这边的写入配置没和ES的能力匹配上。结合你多节点、每个节点跑5-6个spark-submit的场景，咱们一步步来拆解解决：

一、先理解报错本质 #

这个报错是Elasticsearch Spark连接器返回的，意思是某一批次的128条文档里有13条没写成功——ES因为过载触发了限流机制（比如写入线程池队列满、磁盘IO跟不上、内存不足），直接拒绝了部分写入请求。

二、具体优化方案 #

1. 调整Spark侧的写入配置，降低并发压力 #

• 控制单批次写入量：默认的批次大小（es.batch.size.entries=1000）对高并发场景来说可能太大，建议下调到500甚至300，同时限制单批次字节数：

  .option("es.batch.size.entries", "500")
  .option("es.batch.size.bytes", "524288") // 512KB
  

• 减少节点上的spark-submit数量：每个节点跑5-6个独立的提交任务，相当于把写入并发放大了数倍，很容易打满ES。建议合并同类型的任务，或者把每个节点的提交数降到2-3个，甚至错开时间执行。
• 添加自动重试机制：让连接器对失败的批次自动重试，避免直接报错：

  .option("es.batch.write.retry.count", "3")
  .option("es.batch.write.retry.wait", "1000") // 每次重试间隔1秒
  

• 控制并发请求数：限制同时发送给ES的请求数，避免瞬间压垮ES：

  .option("es.batch.concurrent.requests", "2")
  

2. 优化Elasticsearch的写入能力 #

• 临时关闭副本写入：如果是一次性批量导入，可以先把索引的副本数设为0，写完后再改回原来的数值，这样能大幅减少ES的写入开销：

  # 通过ES API修改
  PUT /_all/_settings
  {
    "index.number_of_replicas": 0
  }
  

• 调整写入线程池：ES的写入线程池默认队列是200，线程数等于CPU核心数。如果是专用的ES集群，可以适当调大队列大小（比如改成500），但要注意不要超过JVM内存的承受范围。
• 确保ES资源充足：检查ES节点的磁盘使用率（不要超过85%，否则ES会自动限流）、堆内存（建议是物理内存的一半，最大不超过32G），如果条件允许换成SSD磁盘。

3. 监控排查定位瓶颈 #

• 查看ES的监控面板（比如Kibana），重点看写入吞吐量、写入线程池队列长度、磁盘IO使用率、堆内存使用率，找到具体是哪个资源瓶颈导致的过载。
• 查看ES节点的日志，里面会有更详细的拒绝原因（比如是磁盘满了还是线程池队列溢出），针对性解决问题。

优化后的代码示例 #

input.write
  .format("org.elasticsearch.spark.sql")
  .mode(SaveMode.Append)
  .option("es.resource", "{date}/" + type)
  // 调整批次大小
  .option("es.batch.size.entries", "500")
  .option("es.batch.size.bytes", "524288")
  // 重试配置
  .option("es.batch.write.retry.count", "3")
  .option("es.batch.write.retry.wait", "1000")
  // 控制并发请求
  .option("es.batch.concurrent.requests", "2")
  .save()

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