数据流/波束模板、产品化、初始化和ValueProviders

Question

目前，它需要一整堆必需的可变参数：它需要在哪个Google Cloud项目中运行，它将在哪个bucket和目录前缀中存储文件，它需要从哪个pub/sub订阅中读取，等等。在pipeline.run被称为pipeline.run之前，它对这些参数做了一些工作-验证、字符串拆分等等。在当前的形式中，为了启动作业，我们一直将这些参数传递给PipelineOptionsFactory，并每次都发出新的编译，但似乎应该有更好的方法。我已经将参数设置为ValueProvider对象，但是由于它们是在pipeline.run之外调用的，Maven在编译时抱怨ValueProvider.get（）是在运行时上下文之外调用的（是的，确实如此）

我尝试使用NestedValueProviders，就像在Google“Creating Templates”文档中一样，但是如果我尝试使用NestedValueProvider.of返回文档中所示的字符串，我的IDE会抱怨。我能够编译NestedValueProviders的唯一方法如下：

NestedValueProvider<String, String> pid = NestedValueProvider.of(
        pipelineOptions.getDataflowProjectId(),
        (SerializableFunction<String, String>) s -> s
);

（String pid=nestedValueProvider.of（...）导致以下错误：“不兼容的类型：不存在类型变量T,X的实例，因此org.apache.beam.sdk.options.valueProvider.NestedValueProvider符合java.lang.String”）

我的PipelineOptions中有以下内容：

ValueProvider<String> getDataflowProjectId();
void setDataflowProjectId(ValueProvider<String> value);

由于我们要处理的消息量很大，在管道的前面为通过的每个消息添加这些检查并不实际；我们会很快达到这些电话的每日账户管理限制。

Answer 1

按照当前模板的实现方式，没有必要执行“模板创建后”，而是执行“管道启动前”初始化/验证。

所有现有的验证都将在模板创建期间执行。如果验证检测到这些值不可用（由于是ValueProvider)，则跳过验证。

在某些情况下，可以通过添加运行时检查来近似验证，既可以作为自定义源的初始拆分的一部分，也可以作为DOFN的@setup方法的一部分。在后一种情况下，@setup方法将为创建的DoFn的每个实例运行一次。如果管道是批处理的，则在特定实例失败4次后，管道将失败。