弹性搜索无法写入所有条目：可能是es被重载

Question

我有一个应用程序，我在其中读取 csv 文件并进行一些转换，然后将它们从 spark 本身推送到弹性搜索。像这样

input.write.format("org.elasticsearch.spark.sql")
              .mode(SaveMode.Append)
              .option("es.resource", "{date}/" + type).save()

我有几个节点，在每个节点中，我运行 5-6 个spark-submit推送到elasticsearch

我经常收到错误

Could not write all entries [13/128] (Maybe ES was overloaded?). Error sample (first [5] error messages):
        rejected execution of org.elasticsearch.transport.TransportService$7@32e6f8f8 on EsThreadPoolExecutor[bulk, queue capacity = 200, org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@4448a084[Running, pool size = 4, active threads = 4, queued tasks = 200, completed tasks = 451515]]

我的 Elasticsearch 集群有以下统计信息 -

Nodes - 9 (1TB space,
Ram >= 15GB ) More than 8 cores per node

我修改了elasticseach的以下参数

spark.es.batch.size.bytes=5000000
spark.es.batch.size.entries=5000
spark.es.batch.write.refresh=false

任何人都可以建议，我可以修复什么来摆脱这些错误？

Answer 1

发生这种情况是因为批量请求的传入速率大于 elasticsearch 集群可以处理的速率，并且批量请求队列已满。

默认批量队列大小为 200。

理想情况下，您应该在客户端处理此问题：

1) 通过减少同时运行的 spark-submit 命令的数量

2）通过调整es.batch.write.retry.count和 在拒绝的情况下重试es.batch.write.retry.wait

例子：

es.batch.write.retry.wait = "60s"
es.batch.write.retry.count = 6

在 elasticsearch 集群端：

1) 检查每个索引是否有太多分片并尝试减少它。
该博客对调整分片数量的标准进行了很好的讨论。

2）作为最后的手段增加thread_pool.index.bulk.queue_size

通过对批量拒绝的广泛讨论，查看此博客。