测试Beam PipelineOptions ValueProvider的@默认值

Question

我想要一个数据流模板，其中一个PipelineOptions参数的默认值。

受在线示例的启发，我在PipelineOptions“sub”-界面中使用ValueProvider占位符来进行延迟参数设置：

  @Default.String("MyDefaultValue")
  ValueProvider<String> getMyValue();
  void setMyValue(ValueProvider<String> value);

如果我在运行时指定参数，模板就可以启动一个真正的GCP数据流作业。但是，如果我尝试在真正执行此操作之前不包含参数进行测试：

@Rule
public TestPipeline pipeline = TestPipeline.create();
...
  
@Test
public void test() {
  PipelineOptions options = PipelineOptionsFactory.fromArgs(new String[] {...}).withValidation();
  ...
  pipeline.run(options);
}

然后，当我的TestPipeline执行一个需要参数的DoFn processElement方法时，我得到

IllegalStateException: Value only available at runtime, but accessed from a non-runtime context: 
RuntimeValueProvider{propertyName=myValue, default=MyDefaultValue}
...

更具体地说，它在org.apache.beam.sdk.options.valueProvider中失败：

@Override
public T get() {
  PipelineOptions options = optionsMap.get(optionsId);
  if (options == null) {
    throw new IllegalStateException(...

人们认为运行时是管道运行的时候，这可能是可以原谅的。

Answer 1

当我从Eclipse生成数据流模板时，我也遇到了同样的问题，我的数据流模板从Cloud Composer DAG收到了一个参数。