测试包含 ValueProviders 的管道答案

【问题标题】：Testing a Pipeline containing ValueProviders测试包含 ValueProviders 的管道
【发布时间】：2018-04-06 12:47:36
【问题描述】：

我使用ValueProviders 创建了一个管道，以便将其用作模板。但是，我不知道在测试管道时如何使用ValueProviders。我不能直接使用值来测试，因为我的 PTransforms 正在等待 ValueProviders。

【问题讨论】：

标签： python google-cloud-dataflow apache-beam

【解决方案1】：

我不了解 Python，但在 Java 中您可以使用 StaticValue 提供程序。例如。如果你有如下的选项界面：

interface BaseOptions extends DataflowPipelineOptions {
    void setSource(ValueProvider<String> source);
    ValueProvider<String> getSource();
}

然后您可以使用ValueProvider.StaticValueProvider.of(...) 来初始化您的参数。像这样的：

BaseOptions options = PipelineOptionsFactory.fromArgs(args).as(BaseOptions.class);
options.setSource(ValueProvider.StaticValueProvider.of("/path/to/file"));
Pipeline p = Pipeline.create(options);
p.apply(TextIO.read().from(options.getSource()))
        .apply("just print",
               new ParDo().of(new DoFn<String, String>() {
               @ProcessElement
           public void processElement(ProcessContext c) {
               System.out.println(c.element());

           }
}));
p.run();

【讨论】：

【解决方案2】：

我将Value Providers 的默认值放在我的管道选项中：

class MypipelineOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls,parser):
      parser.add_value_provider_argument('--variable',
                                        type=float,
                                        dest='variable',
                                        default=5)

这样当我在测试文件中调用 MyPipelineOptions 时，它会自动使用默认值。

【讨论】：