参数从 rest api 传递到数据流作业时出错 - 云函数答案

【问题标题】：Error with Argument passing to dataflow job fro rest api - cloud function参数从 rest api 传递到数据流作业时出错 - 云函数
【发布时间】：2020-08-06 18:00:58
【问题描述】：

任何人都可以分享数据流 python 代码来接受参数吗？通过rest API传递的参数我面临同样的问题。我的 df 代码如下：-

def run(argv=None):

    parser = argparse.ArgumentParser()

    # Specifically we have the input file in CSV format to read and the output BQ table to write.
    # This is the final stage of the pipeline, where we define the destination

    parser.add_argument(
        '--input',
        dest='input',
        required=False,
        help='Input file to read. This can be a local file or '
        'a file in a Google Storage Bucket.',
        # This example file contains a total of only 10 lines.
        # Useful for developing on a small set of data.
        default='gs://intient_output/measurementunit.csv')

    parser.add_argument(
        '--output',
        dest='output',
        required=False,
        help='Output file to be written. This can be a local file or '
        'a file in a Google Storage Bucket.',
        default='mygcpdataengineerlab:intientpoc.measurementunit'
        )

    # Parse arguments from the command line.
    known_args, pipeline_args = parser.parse_known_args(argv)
    data_ingestion = DataIngestion()
    project = ''

    p = beam.Pipeline(options=PipelineOptions(pipeline_args))

下面的异常堆栈跟踪

Error-    response = request.execute()
  File "/env/local/lib/python3.7/site-packages/googleapiclient/_helpers.py", line 134, in positional_wrapper
    return wrapped(*args, **kwargs)
  File "/env/local/lib/python3.7/site-packages/googleapiclient/http.py", line 907, in execute
    raise HttpError(resp, content, uri=self.uri)
googleapiclient.errors.HttpError: <HttpError 400 when requesting https://dataflow.googleapis.com/v1b3/projects/mygcpdataengineerlab/templates:launch?gcsPath=gs%3A%2F%2Fgcp_dataflow_csv_bq_code%2Ftemplates&alt=json returned "(9744cfd1809f74a): The workflow could not be created. Causes: (9744cfd1809fa2d): Found unexpected parameters: ['input' (perhaps you meant 'update'), 'output' (perhaps you meant 'job_port')]">

【问题讨论】：

确保将 --input 和 --output 传递给正确的程序。发生此错误是因为您必须以某种方式将 --input 和 --output 传递给 Dataflow 服务。 Dataflow 本身无法识别 --input 和 --output，只有 Python 脚本中的自定义解析器可以识别这些选项。如果您仍然遇到问题，可以提供有关如何运行作业的更多信息。
我正在通过云功能运行代码如下：-
template_body = { "jobName": job, "parameters": { "input": "gs://input/abc.csv", "output": "project:dataset.table", }, "环境": { "tempLocation": "gs://gcp_dataflow_csv_bq_code/temp", } } request = service.projects().templates().launch(projectId=gcp_project, gcsPath=template_path, body=template_body) 响应= request.execute() 当我删除参数选项时，df 被默认选项触发
我还创建了一个 _metadata 文件 { "description": "Dataflow job template to load data from csv to BQ", "name": "parameters", "parameters": [ "name": " input", "name": "output", ] } 根据链接harness.io/2019/10/…
我明白了。您的代码与您提供的示例之间的区别在于您直接使用 argparse，而示例正确使用了 Beam 的参数解析器。您将需要创建 Beam 的 PipelineOptions 的子类来定义您的自定义参数，例如他们示例中的 UserOptions。

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

【解决方案1】：

问题在于dest=input，因为最新的光束选项不支持此关键字参数。为简单起见，您可以执行以下操作：

def run(argv=None, save_main_session=True):
  """Build and run the pipeline."""
  parser = argparse.ArgumentParser()
  parser.add_argument(
      '--output_topic', required=True,
      help=('Output PubSub topic of the form '
            '"projects/<PROJECT>/topics/<TOPIC>".'))

  parser.add_argument(
      '--input_topic',
      help=('Input PubSub topic of the form '
            '"projects/<PROJECT>/topics/<TOPIC>".'))

  known_args, pipeline_args = parser.parse_known_args(argv)
  pipeline_options = PipelineOptions(pipeline_args)
  pipeline_options.view_as(SetupOptions).save_main_session = save_main_session

【讨论】：

像 OP 代码中的 --input 和 --output 一样，--output_topic 和 --input_topic 是 Beam/Dataflow 无法识别的本地定义参数。

【解决方案2】：

我通过接受 arg 作为管道选项来解决它。 MyPipeOpt 类（管道选项）： @classmethod def _add_argparse_args(cls, 解析器): parser.add_value_provider_argument('--input', help='输入要读取的文件。这可以是本地文件或 Google 存储桶中的文件。'）

pipeline_options = PipelineOptions(argv) my_options = pipeline_options.view_as(MyPipeOpt)

p = beam.Pipeline(options=pipeline_options) 并将变量用作创建的光束管道中的 my_options.input。

谢谢，

【讨论】：

您介意添加格式并显示您放置所描述的 sn-p 的位置吗