从 Python 云函数调用数据流弹性模板时出错

Question

我正在尝试使用以下代码从 Python 云函数调用 Dataflow flex 模板：

from googleapiclient.discovery import build

dataflow = build('dataflow', 'v1b3')
request = dataflow.projects().templates().launch(
    projectId=projectid,
    gcsPath=template,
    body={
        'jobName': job_name,
        'parameters': parameters,
    }
)
print(f"Start to execute Dataflow Job with name: {job_name}")
return request.execute()

如果引用经典模板，同样的代码可以工作，但引用弹性模板时，它会出错，并显示以下详细信息：

Traceback (most recent call last): File "/env/local/lib/python3.7/site-packages/google/cloud/functions/worker_v2.py", line 449, in run_background_function _function_handler.invoke_user_function(event_object) File "/env/local/lib/python3.7/site-packages/google/cloud/functions/worker_v2.py", line 268, in invoke_user_function return call_user_function(request_or_event) File "/env/local/lib/python3.7/site-packages/google/cloud/functions/worker_v2.py", line 265, in call_user_function event_context.Context(**request_or_event.context)) File "/user_code/main.py", line 41, in trigger_cleaning response = call_dataflow(project_id, job_name, template, parameters) File "/user_code/utils/gcp_utils.py", line 58, in call_dataflow 'parameters': parameters, 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 915, in execute raise HttpError(resp, content, uri=self.uri) googleapiclient.errors.HttpError: <HttpError 400 when requesting https://dataflow.googleapis.com/v1b3/projects/concise-flame-279117/templates:launch?gcsPath=gs%3A%2F%2Filan-artefacts%2Ftemplates%2Ftemplate666.json&alt=json returned "(145ec02dfb186de2): There is no support for job type with environment version . Please try upgrading the SDK to the latest version. You can find the instructions on installing the latest SDK at https://cloud.google.com/dataflow/docs/guides/installing-beam-sdk. If that doesn't work, please contact the Cloud Dataflow team for assistance at https://cloud.google.com/dataflow/support.

那个“环境版本不支持作业类型。请尝试将SDK升级到最新版本。您可以在https://cloud.google.com/dataflow/docs/guides/installing-beam-sdk找到安装最新SDK的说明。如果这不起作用，请通过 https://cloud.google.com/dataflow/support" 联系 Cloud Dataflow 团队寻求帮助，错误没有任何意义，因为这是在 Cloud 函数中运行...

有什么想法吗？提前致谢！

更新：已解决

下面的代码解决了这个问题：

dataflow = build('dataflow', 'v1b3')
request = dataflow.projects().locations().flexTemplates().launch(
    projectId=projectid,
    location=location,
    body={
        'launchParameter': {
            'jobName': job_name,
            'containerSpecGcsPath': template,
            'parameters': parameters
        }
    }
)
print(f"Start to execute Dataflow Job with name: {job_name}")
return request.execute()

Answer 1

嗯，看起来有一个单独的端点/调用来运行弹性模板：docs。这对我有用：

dataflow = build('dataflow', 'v1b3', cache_discovery=False)
request = dataflow.projects().locations().flexTemplates().launch(
    projectId='my-project',
    location='us-central1',
    body={
      'launch_parameter': {
        'jobName': 'testing-1234',
        'parameters': {
          'some_parameter': 'hello',
        },
        'containerSpecGcsPath': 'gs://my-storage-bucket/flex_templates/do_something.json'
      }
    }
)

Answer 2

下面的代码解决了这个问题：

dataflow = build('dataflow', 'v1b3')
request = dataflow.projects().locations().flexTemplates().launch(
    projectId=projectid,
    location=location,
    body={
        'launchParameter': {
            'jobName': job_name,
            'containerSpecGcsPath': template,
            'parameters': parameters
        }
    }
)
print(f"Start to execute Dataflow Job with name: {job_name}")
return request.execute()