Apache Beam Elastic IO模块是否支持按ID更新Elasticsearch已有文档？

Apache Beam的Elasticsearch IO模块支持按ID更新已存在的Elasticsearch文档，具体实现方式如下：

• 指定更新操作类型：在配置ElasticsearchIO.Write时，通过withOperationType()方法将操作类型设置为ElasticsearchIO.OperationType.UPDATE，IO模块会根据文档ID匹配目标资源执行更新逻辑。
• 绑定文档ID：确保你的数据元素中包含Elasticsearch文档的_id字段，并通过withIdFn()函数定义如何从元素中提取该ID，IO会以此ID定位要更新的文档。
• Upsert逻辑支持：如果需要实现「存在则更新，不存在则插入」的Upsert操作，可以启用withUpsert(true)，无需额外区分操作场景。

示例代码（Java） #

Pipeline pipeline = Pipeline.create(pipelineOptions);

pipeline.apply("Read Source Data", /* 你的数据源读取逻辑 */)
        .apply("Write to Elasticsearch", ElasticsearchIO.write()
                .withConnectionConfiguration(ElasticsearchIO.ConnectionConfiguration.create(
                        new String[]{"http://es-host:9200"},
                        "target-index",
                        "doc-type"))
                .withOperationType(ElasticsearchIO.OperationType.UPDATE)
                .withIdFn(record -> record.getDocumentId()) // 自定义ID提取逻辑
                .withUpsert(true)); // 可选：启用Upsert模式

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