使用 REST API 创建管道会导致错误 400：INVALID_ARGUMENT

Question

我正在使用 Python 直接调用 Dataflow REST API (https://cloud.google.com/dataflow/docs/reference/data-pipelines/rest/v1/projects.locations.pipelines/create)请求模块创建管道。我检索了现有工作管道的定义作为基础，并对其进行了修改以创建新的不同管道。我已经仔细检查了所有参数，但最终还是

{'error': {'code': 400, 'message': 'Request contains an invalid argument.', 'status': 'INVALID_ARGUMENT'}}

API 没有说明哪个参数无效。出了什么问题？我想为 Python 使用正确的客户端库，但对于 Dataflow，它仍处于预览阶段，不支持管道创建。

这是我的代码：

body = {'name': 'projects/<project_id>/locations/europe-west1/pipelines/test_name',
 'displayName': 'test_name',
 'type': 'PIPELINE_TYPE_BATCH',
 'state': 'STATE_ACTIVE',
 'workload': {'dataflowFlexTemplateRequest': {'projectId': '<project_id>',
   'launchParameter': {'jobName': 'job_test2',
    'parameters': {'password': '<base64 encoded string>',
     'isTruncate': 'true',
     'useColumnAlias': 'true',
     'serviceAccount': '[email protected]',
     'experiments': 'use_runner_v2',
     'driverClassName': 'com.amazon.redshift.jdbc.Driver',
     'connectionProperties': 'autosave=never',
     'workerMachineType': 'n2-highmem-4',
     'bigQueryLoadingTemporaryDirectory': 'gs://censored-temp',
     'connectionURL': '<base64 encoded string>',
     'defaultWorkerLogLevel': 'DEBUG',
     'maxNumWorkers': '2',
     'query': 'select column1 from schema_name.table_name',
     'driverJars': 'gs://censored-assets/redshift-jdbc42-2.1.0.10.jar',
     'username': '<base64 encoded string>',
     'outputTable': '<project_id>:<dataset_name.<table_name>'},
    'containerSpecGcsPath': 'gs://dataflow-templates-europe-west1/latest/flex/Jdbc_to_BigQuery_Flex',
    'environment': {'workerZone': 'europe-west4-a',
     'kmsKeyName': 'projects/<project_id>/locations/global/keyrings/<keyring_id>/<key_id>'}},
   'location': 'europe-west1'}}}


headers = { 'Authorization': f'Bearer {creds.token}',
            'Content-Type': 'application/json; charset=UTF-8'}
response = requests.post(url_create, json=body, headers=headers)

结果：

{'error': {'code': 400, 'message': 'Request contains an invalid argument.', 'status': 'INVALID_ARGUMENT'}}

我已经仔细检查了所有参数值是否正确，并且我还尝试删除和更改一些参数值，看看是否可以查明哪个参数值有问题，但无济于事。

Answer 1

您似乎正在尝试启动模板作业。根据 https://cloud.google.com/dataflow/docs/concepts/dataflow-templates#compare-flex-templates-and-classic-templates ，你应该使用

https://cloud.google.com/dataflow/docs/reference/rest/v1b3/projects.locations.flexTemplates/launch

google-cloud-dataflow-client 提供 REST 客户端并具有示例：https://github.com/googleapis/google-cloud-python/tree/main/packages/google-cloud-dataflow-client/samples/生成的样本