无法在我的数据流管道中创建具有值覆盖参数的经典模板答案

【问题标题】：Unable to create classic template with value override parameter in my dataflow pipeline无法在我的数据流管道中创建具有值覆盖参数的经典模板
【发布时间】：2021-08-30 01:31:16
【问题描述】：

我的场景是我正在调用具有云功能的数据流管道，我创建了经典模板并成功调用它。但是我的管道包含在我的代码中编码的 GCS 输入路径。

我想在运行时将它与 API 作为参数传递：{input:gcspath} 在请求正文中。为了做到这一点我在 DOC 中搜索它告诉我需要覆盖 value 参数，并且我在我的代码中覆盖 valueparameter 但是当我创建模板时没有创建模板。可能是我的代码错误如果有人可以检查我错过了其中的某些内容，我将粘贴下面的代码。或者如果有人可以粘贴小的正确样本 sn-p。

------------------我的代码-------------

import apache_beam as beam
# from apache_beam.io import fileio
import argparse
from apache_beam.options.pipeline_options import PipelineOptions
from sys import argv 


PROJECT_ID = 'xxxxx-rnd'
# SCHEMA = 'sr:INTEGER,abv:FLOAT,id:INTEGER,name:STRING,style:STRING,ounces:FLOAT'
SCHEMA = 'High:STRING,Low:STRING,Open:STRING,Close:STRING,Volume:STRING,AdjClose:STRING'

def discard_incomplete(data):
    """Filters out records that don't have an information."""
    return len(data['abv']) > 0 and len(data['id']) > 0 and len(data['name']) > 0 and len(data['style']) > 0

def convert_types(data):
    """Converts string values to their appropriate type."""
    data['Date'] = str(data['Date']) if 'Date' in data else None
    data['High'] = str(data['High']) if 'High' in data else None
    data['Low'] = str(data['Low']) if 'Low' in data else None
    data['Open'] = str(data['Open']) if 'Open' in data else None
    data['Close'] = str(data['Close']) if 'Close' in data else None
    data['Volume'] = str(data['Volume']) if 'Volume' in data else None
    data['AdjClose'] = str(data['AdjClose']) if 'AdjClose' in data else None
    # data['Average'] = float(data['Average']) if 'Average' in data else None
    return data


def del_unwanted_cols(data):
    """Delete the unwanted columns"""
    del data['Date']
    # del data['brewery_id']
    return data
 
class WordcountOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
        # Use add_value_provider_argument for arguments to be templatable
        # Use add_argument as usual for non-templatable arguments
        parser.add_value_provider_argument(
            '--input',
            default='gs://xxx/convertcsv.csv',
            help='Path of the file to read from')
        parser.add_argument(
            '--output',
            required=True,
            help='Output file to write results to.')

pipeline_options = PipelineOptions(['--output', 'gs://xxx/results/output'],
    runner='DataflowRunner',
    project='rightmechanics-rnd',
    job_name='dataflow-intro',
    temp_location='gs://xxx/valuetemp',
    region='us-central1'
    )
p = beam.Pipeline(options=pipeline_options)
wordcount_options = pipeline_options.view_as(WordcountOptions)
(p  | 'ReadData' >> beam.io.ReadFromText(wordcount_options.input, skip_header_lines =1)
    | 'SplitData' >> beam.Map(lambda x: x.split(','))
    | 'FormatToDict' >> beam.Map(lambda x: {"Date": x[0], "High": x[1], "Low": x[2], "Open": x[3], "Close": x[4], "Volume": x[5], "AdjClose": x[6]}) 
    #    | 'DeleteIncompleteData' >> beam.Filter(discard_incomplete)
    | 'ChangeDataType' >> beam.Map(convert_types)
    | 'DeleteUnwantedData' >> beam.Map(del_unwanted_cols)
    #    | 'WriteToStorageBucket' >> beam.io.fileio.WriteToFiles(
    #        '{0}:beer.beer_data'.format(PROJECT_ID),
    #        schema=SCHEMA,
    #        write_disposition=beam.io.fileio.WRITE_APPEND
    #    )
    | 'WriteToBigQuery' >> beam.io.WriteToBigQuery(
        '{0}:beer.beer_data2'.format(PROJECT_ID),
        schema=SCHEMA,
        write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
        method='FILE_LOADS'
        )
    )
result = p.run()
result.wait_until_finish()

【问题讨论】：

标签： google-cloud-platform google-cloud-functions google-cloud-dataflow

【解决方案1】：

您可以找到创建经典模板here的说明和示例。

它说

编写管道后，您必须创建和暂存模板文件。

会不会是你创建和暂存模板文件的cmd不正确？

创建并暂存模板后，下一步是执行模板。

【讨论】：