是否Elasticsearch chIO为apache束java支持模板和ValueProvider参数？调用模板时出错

Question

我试图创建一个模板Apache束索引数据到elasticsearch。正在创建模板，但是在调用模板时管道失败，没有协议错误。看起来很奇怪，因为错误与URL对象有关。

public class StarterPipeline {
    private static final Logger LOG = LoggerFactory.getLogger(StarterPipeline.class);

    public interface IndexToEsOptions extends PipelineOptions {
        @Description("Path of the gzip index file to read from")
        ValueProvider<String> getInputIndexFile();
        void setInputIndexFile(ValueProvider<String> value);

        @Description("Index name to index with")
        ValueProvider<String> getIndexName();
        void setIndexName(ValueProvider<String> value);

        @Description("Index template name")
        ValueProvider<String> getIndexTemplate();
        void setIndexTemplate(ValueProvider<String> value);

        @Description("URI for es")
        @Default.String("https://vpc-myescore01-5mtib6vgjw7sbhgn3kbnwnluim.us-west-1.es.amazonaws.com")
        ValueProvider<String> getEsUri();
        void setEsUri(ValueProvider<String> value);
    }

    public static void main(String[] args) {



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

        Pipeline p = Pipeline.create(options);
        p.apply(TextIO.read().from(options.getInputIndexFile()))
                .apply(ElasticsearchIO.write().withConnectionConfiguration(
                        ElasticsearchIO.ConnectionConfiguration.create(
                                new String[]{options.getEsUri().toString()},
                                options.getIndexName().toString(),
                                options.getIndexTemplate().toString())
                                .withConnectTimeout(240)
                        )
                        .withMaxBatchSizeBytes(15 * 1024 * 1024)
                );

        p.run();
    }

我运行时遇到的错误是

Answer 1

简单地说，不，它看起来不像ElasticsearchIO。ConnectionConfiguration支持ValueProviders，至少在当前版本（2.22.0）中不支持。通过查看ConnectionConfiguration的签名可以看到这一点。创建：

public static ElasticsearchIO.ConnectionConfiguration create(java.lang.String[] addresses,
                                                             java.lang.String index,
                                                             java.lang.String type)

并将其与支持ValueProviders的函数进行比较，ElasticsearchIO。阅读withQuery：

public ElasticsearchIO.Read withQuery(ValueProvider<java.lang.String> query)

要支持ValueProviders，函数必须实际接受ValueProvider对象。这是因为ValueProvider打算在运行时而不是在管道构建期间通信参数。因此，在管道构建期间，它必须作为ValueProvider对象到处传递。

在您的示例中，您正在调用EsUri的ValueProvider上的toString，而不是生成包含URL的字符串，而是得到ValueProvider的字符串表示形式，如下所示：“RuntimeValueProvider{propertyName=EsUri，默认值=https://vpc-esprdcore01-5mtib6vgjw7sbhgn3kbnwnluim.us-east-1.es.amazonaws.com}。这就是为什么会出现格式错误的异常异常。它试图将该字符串作为URL读取，但失败。

解决方案很简单，您只需要将EsUri参数更改为构建时间参数，方法是从ValueProvider更改