如何从DataFlow模板中提取REST API参数？

Question

我想在Google Cloud平台中使用Cloud Function触发数据流。

在Cloud Function中，我使用Python这样调用API：

from googleapiclient.discovery import build

service = build('dataflow', 'v1b3', cache_discovery=False)

request = service.projects().locations().templates().launch(
    projectId=projectId, gcsPath=srcDataFlowTemplate, location='us', body={
        'jobName': jobName,
        'parameters': parameters,
        'environment':environment
    }
)

response = request.execute()

这是我正在启动的数据流模板中的管道：

(p
   | 'Create PCollection' >> beam.Create(inputFile)
   | 'Read from a File' >> beam.io.ReadAllFromText(skip_header_lines=1)
   | 'Write to BigQuery' >> beam.io.Write(beam.io.WriteToBigQuery (table,
                project=projectId,
                dataset=dataset,
                schema=schema,
                create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED,
                write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND
        )
    )
)

在数据流模板中-是否可以从应用程序接口调用中提取此'inputFile‘，它存储在请求主体中的parameters属性和projectId下？

Answer 1

在具有光束框架的数据流上运行作业时，可以使用DataflowRunner。当您创建管道时，您使用"options“对象创建它。

p = beam.Pipeline(options=options)

This object contains basic parameters，例如projectId，您可以对其进行扩展以添加您自己的管道参数。