【问题标题】:Dataflow template not using the runtime parameters未使用运行时参数的数据流模板
【发布时间】:2019-05-23 17:53:09
【问题描述】:

我正在使用数据流模板来运行云数据流

我提供了一些默认值和调用模板。 Dataflow 在数据流管道摘要中正确显示管道选项。但它不采用运行时值。

class Mypipeoptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
        parser.add_value_provider_argument(
            '--preprocess_indir',
            help='GCS path of the data to be preprocessed',
            required=False,
            default='gs://default/dataset/'
        )
        parser.add_value_provider_argument(
            '--output_dir_train',
            help='GCS path of the preprocessed train data',
            required=False,
            default='gs://default/train/'
        )
        parser.add_value_provider_argument(
            '--output_dir_test',
            help='GCS path of the preprocessed test data',
            required=False,
            default='gs://default/test/'
        )
        parser.add_value_provider_argument(
            '--output_dir_validate',
            help='GCS path of the preprocessed validate data',
            required=False,
            default='gs://default/validate/'
        )

然后我正在检查值是否可以访问

p = beam.Pipeline(options=args)

    if args.preprocess_indir.is_accessible():
        input_dir = args.preprocess_indir
    else:
        input_dir = getValObj(args.preprocess_indir)


    if args.output_dir_train.is_accessible():
        output_train = args.output_dir_train
    else:
        output_train = getValObj(args.output_dir_train)

    if args.output_dir_test.is_accessible():
        output_test = args.output_dir_test
    else:
        output_test = getValObj(args.output_dir_test)

    if args.output_dir_validate.is_accessible():
        output_validate = args.output_dir_validate
    else:
        output_validate = getValObj(args.output_dir_validate)

现在在调用模板时,我可以看到我想要作为 (Mypipeoptions)pipeine 选项参数传递的值,但它没有在实际运行中使用,而是使用给定的默认选项

【问题讨论】:

  • 您将选项传递给哪些转换,哪些选项不使用运行时值?
  • 无,如图所示,这些选项用于 text.io 读取和 tfrecord.io 下沉。
  • 您能否更新部分问题以包括您如何在管道中定义转换?

标签: google-cloud-dataflow apache-beam


【解决方案1】:

我想我找到了解决方案,我将运行时参数分配给变量,然后将其传递给输入或输出。

当我直接将运行时参数传递给源或接收器时,它起作用了。像下面那个

 'Write train dataset to destination' >> beam.io.tfrecordio.WriteToTFRecord(
        file_path_prefix=args.output_dir_train
    ) 

我相信我错过的部分是,在创建模板时,它会构建图形,并且只有运行时参数可以插入其运行时。其他计算在构建图时已经完成。

如有错误请指正

【讨论】:

  • 你的正确。您需要将值提供者对象传递到正在使用它们的转换中,以便在运行时读取它们时,它们可以从 PipeineOptions 获取它们。
  • 你好@user3483651,很抱歉打扰大家,我面临类似的问题(对于 RuntimeValueProvider 也是)但无法绕过。当你有时间时,你能检查一下吗?这是链接:stackoverflow.com/questions/59894035/…
猜你喜欢
  • 1970-01-01
  • 1970-01-01
  • 2020-12-21
  • 2018-05-27
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 2019-01-19
相关资源
最近更新 更多