AWS OpenSearch 的 Apache Beam Java ElasticsearchIO 错误

Question

我正在尝试在 GCP 项目中部署 Apache Beam 批处理管道。我的管道将从 AWS OpenSearch 中的 Elasticsearch 集群中读取数据。

这是我为做同样的事情而编写的代码：

        ElasticSearchOptions options = PipelineOptionsFactory.fromArgs(args).withValidation()
                .as(ElasticSearchOptions.class);

        Pipeline pipeline = Pipeline.create(options);

        options.setAwsCredentialsProvider(
                new AWSStaticCredentialsProvider(new BasicAWSCredentials(
                        options.getAwsAccessKey(),
                        options.getAwsSecretKey()
                )
            )
        );
        options.setAwsRegion("us-east-1");

        PCollection<String> output =
            pipeline.apply(
                    ElasticsearchIO.read().withConnectionConfiguration(
                            ElasticsearchIO.ConnectionConfiguration.create(
                                    <hostName>,
                                    options.getElasticSearchIndex(),
                                    options.getElasticSearchType()
                                    )
                    )
                .withQuery("field_name:some-value")
            );

我遇到了一个错误

403: User anonymous is not authorized to perform: es:ESHTTPGet

为了更好地理解这个错误，我从我的工作节点打印了 AWS 密钥和访问密钥，以确保传递的值得到正确处理。确实如此！

我被困在这一点上。如何使用我的 AWS 凭证（访问和密钥）授权自己并签署我对 Elasticsearch Read 运算符的请求？

Answer 1

我可能猜想 ElasticsearchIO 对 AWS 凭证和您之前设置的 AwsCredentialsProvider 一无所知，它仅使用您传递给 .withConnectionConfiguration() 的 ConnectionConfiguration 实例来连接到集群。

因此，如果 ConnectionConfiguration withUsername(String) 和 ConnectionConfiguration withPassword(String) 与 AWS 凭证相同，您可以尝试设置您的用户名/密码（tbh，我不知道 Elasticsearch 身份验证如何与 AWS OpenSearch 配合使用）。