在Google Cloud Dataflow-Python中访问模板化运行时参数

Question

我正在尝试为Google Cloud Dataflow创建自己的模板，这样作业就可以从GUI执行，让其他人更容易执行。我遵循了教程，创建了自己的PipelineOptions类，并用parser.add_value_provider_argument()方法填充了它。然后，当我尝试使用my_options.argname.get（）将这些参数传递到管道中时，我会得到一个错误，告诉我该项不是从运行时上下文调用的。我不明白这个。Arg不是定义管道图本身的一部分，它们只是参数，如输入文件名、输出表名等。

下面是代码。如果我对输入文件名、输出表名、写处理和分隔符进行硬编码，它就可以工作。如果我用它们的my_options.argname.get（）等效项替换它们，则失败。在所示的代码片段中，除了outputBQTable名称之外，我已经对所有内容进行了硬编码，在该名称中我使用my_options.outputBQTable.get（）。此操作失败，出现以下消息。

apache_beam.error.runtimeValueProviderError：RuntimeValueProvider（选项：outputBQTable,type：str,default_value:'dataflow_csv_reader_testing.names').get（）未从运行时上下文调用

我很感激任何关于如何让它发挥作用的指导。

import apache_beam
from apache_beam.io.gcp.gcsio import GcsIO
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.value_provider import RuntimeValueProvider
import csv
import argparse

class MyOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls,parser):
        parser.add_value_provider_argument('--inputGCS', type=str,
              default='gs://mybucket/df-python-csv-test/test-dict.csv',
              help='Input gcs csv file, full path and filename')
        parser.add_value_provider_argument('--delimiter', type=str,
              default=',',
              help='Character used as delimiter in csv file, default is ,')
        parser.add_value_provider_argument('--outputBQTable', type=str,
              default='dataflow_csv_reader_testing.names',
              help='Output BQ Dataset.Table to write to')
        parser.add_value_provider_argument('--writeDisposition', type=str,
              default='WRITE_APPEND',
              help='BQ write disposition, WRITE_TRUNCATE or WRITE_APPEND or WRITE_EMPTY')

def main():
    optlist=PipelineOptions()
    my_options=optlist.view_as(MyOptions)
    p = apache_beam.Pipeline(options=optlist)
    (p
    | 'create'            >> apache_beam.Create(['gs://mybucket/df-python-csv-test/test-dict.csv'])
    | 'read gcs csv dict' >> apache_beam.FlatMap(lambda file: csv.DictReader(apache_beam.io.gcp.gcsio.GcsIO().open(file,'r'), delimiter='|'))
    | 'write bq record'   >> apache_beam.io.Write(apache_beam.io.BigQuerySink(my_options.outputBQTable.get(), write_disposition='WRITE_TRUNCATE'))
    )
    p.run()

if __name__ == '__main__':
    main()