使用PubSub to Elasticsearch模板遇ES写入错误的技术咨询

我们运行Google Dataflow批处理作业向PubSub写入数据，同时部署了独立的流作业，通过「PubSub to Elasticsearch」模板从PubSub拉取数据更新Elasticsearch索引。但多次出现Error writing to ES after 1 attempt(s). No more attempts allowed错误，不得不终止流作业。

当前作业写入的是非流索引，且需要多次更新同一文档，咨询以下问题：

1. 是否需要改用流索引？
2. 若无需改用，如何配置Dataflow作业降低写入速率，避免压垮Elasticsearch？已知可增大thread_pool.index.bulk.queue_size但不推荐此方法。

错误堆栈信息：

Error message from worker: generic::unknown: org.apache.beam.sdk.util.UserCodeException: java.io.IOException: Error writing to ES after 1 attempt(s). No more attempts allowed
    org.apache.beam.sdk.util.UserCodeException.wrap(UserCodeException.java:39)
    org.apache.beam.sdk.io.elasticsearch.ElasticsearchIO$BulkIO$BulkIOBundleFn$DoFnInvoker.invokeFinishBundle(Unknown Source)
    org.apache.beam.fn.harness.FnApiDoFnRunner.finishBundle(FnApiDoFnRunner.java:1751)
    org.apache.beam.fn.harness.data.PTransformFunctionRegistry.lambda$register$0(PTransformFunctionRegistry.java:111)
    org.apache.beam.fn.harness.control.ProcessBundleHandler.processBundle(ProcessBundleHandler.java:538)
    org.apache.beam.fn.harness.control.BeamFnControlClient.delegateOnInstructionRequestType(BeamFnControlClient.java:151)
    org.apache.beam.fn.harness.control.BeamFnControlClient$InboundObserver.lambda$onNext$0(BeamFnControlClient.java:116)
    java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.io.IOException: Error writing to ES after 1 attempt(s). No more attempts allowed
    org.apache.beam.sdk.io.elasticsearch.ElasticsearchIO$BulkIO$BulkIOBaseFn.handleRetry(ElasticsearchIO.java:2569)
    org.apache.beam.sdk.io.elasticsearch.ElasticsearchIO$BulkIO$BulkIOBaseFn.flushBatch(ElasticsearchIO.java:2519)
    org.apache.beam.sdk.io.elasticsearch.ElasticsearchIO$BulkIO$BulkIOBaseFn.flushAndOutputResults(ElasticsearchIO.java:2435)
org.apache.beam.sdk.io.elasticsearch.ElasticsearchIO$BulkIO$BulkIOBaseFn.finishBundle(ElasticsearchIO.java:2396)

1. 是否需要改用流索引？ #

不需要强制改用流索引。流索引（如ILM管理的数据流）核心优势是自动生命周期管理，更适配持续写入的时序类数据，但你的场景是多次更新同一文档——流索引底层仍由普通索引组成，文档更新同样需要定位到具体分片，无法从根源解决写入过载问题。如果你的数据不属于时序类，继续使用非流索引完全可行。

2. 配置Dataflow降低写入速率的方案 #

调整批量写入参数 #

• 减小单批次大小：在Dataflow模板中设置batchSize（单批次文档数）、batchSizeBytes（单批次字节数）为更小值，比如从默认1000条降到200-500条，降低单次请求对ES的压力。
• 延长批次间隔：修改batchInterval参数，延长攒批时间，控制ES接收请求的频率，比如从10秒调整到30秒。

控制作业并行度 #

• 限制Worker数量：部署作业时指定maxNumWorkers参数，减少并发写入的Worker实例，直接降低整体写入QPS。
• 设置数据分区键：给数据设置分区键，让同一文档的更新请求落到同一Worker批次，既减少ES分片竞争，也能通过分区数量间接限制并行写入流数。

优化重试机制 #

• 增加重试次数与间隔：将maxRetryAttempts从默认1次调整为3-5次，同时设置retryDelay延长重试间隔，给ES留出恢复处理能力的时间。注：重试不会导致文档更新冲突，因为目标是同一文档的重复更新。

流量削峰配置 #

• 限制PubSub拉取速率：在Dataflow作业中配置maxOutstandingMessages和maxOutstandingBytes参数，限制Worker从PubSub拉取消息的速率，避免消息堆积引发ES写入流量突增。

优化ES写入逻辑 #

• 启用文档版本控制：每次更新请求携带正确的版本号，避免ES因版本冲突返回错误，减少无效重试和写入压力。

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