Read value provider argument in dataflow template

user_14785843 0
Viewed 417

I have cloud function that launches dataflow template when new file is loaded and passes route on GS to that file. I have to read that as a value provider and pass it in pipeline like this.

 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())))
    # ...
)

But I get this error when template starts: AttributeError("'RuntimeValueProvider' object has no attribute 'strip'")

How do I solve this?

apache-beam google-cloud-dataflow google-cloud-platform
1 Answers

Unfortunately, according to the documentation the MatchFiles method does not seem to take ValueProvider arguments, only strings. A workaround would be to use the MatchAll method instead since it uses a PCollection as input. By creating an Impulse transform outputting to a custom DoFn that uses your ValueProvider you can output a PCollection containing your ValueProvider's value as a single element, which will be read by MatchAll.

user_113370330
Related