0

我试图在使用 ValueProvider 执行运行命令时获取服务器 ID 作为参数

对于选项接口中的值提供者：

ValueProvider<String> getKafkaServer();
    void setKafkaServer(ValueProvider<String> value);

withBootstrapServers 抛出错误“不兼容的类型：org.apache.beam.sdk.options.ValueProvider 无法转换为 java.lang.String”

PCollection<String> kafkaMessages = p
                .apply("Read from Kafka", KafkaIO.<Long, String>read()
                        .withBootstrapServers(options.getBootstrapServers())

这个答案建议使用 options.getBootstrapServers().get() 但这会产生以下错误

[ERROR] - Expected getter for property [getbootstrapServers] of type [org.apache.beam.sdk.options.ValueProvider]
[ERROR] - Expected setter for property [bootstrapServers] of type [org.apache.beam.sdk.options.ValueProvider]

非常感谢任何解决此问题的帮助

4

1 回答 1

0

ValueProviders 仅适用于对其有深入支持的转换（包括源）。只有旧版 Dataflow 模板才需要它们。通常，可以为管道选项（例如String getKafkaServer()）使用具体值，这些值可以在管道启动时传递。如果想将此选择推迟到模板实例化时间，请使用flex templates。

于 2021-10-05T18:59:13.493 回答