读取数据流模板中的值提供程序参数

Question

我有云功能，可以在加载新文件时启动数据流模板并将 GS 上的路由传递给该文件。我必须将其作为价值提供者阅读并像这样在管道中传递它。

 class DataFlowOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
        parser.add_value_provider_argument('--input')

dataflow_options = PipelineOptions().view_as(DataFlowOptions)

pipeline = apache_beam.Pipeline(options = dataflow_options)

(pipeline
    | fileio.MatchFiles(dataflow_options.input)
    | fileio.ReadMatches()
    | apache_beam.FlatMap(lambda f: csv.DictReader(io.TextIOWrapper(f.open())))
    # ...
)

但模板启动时出现此错误：AttributeError("'RuntimeValueProvider' object has no attribute 'strip'")

我该如何解决这个问题？

Answer 1

不幸的是，根据the documentation，MatchFiles 方法似乎不接受 ValueProvider 参数，只有字符串。一种解决方法是改用MatchAll 方法，因为它使用PCollection 作为输入。通过创建Impulse transform 输出到custom DoFn that uses your ValueProvider，您可以将包含ValueProvider 值的PCollection 输出为单个元素，MatchAll 将读取该值。