来自作曲家错误的 gcp 触发数据流作业
gcp trigger dataflow job from composer error
我正在尝试 运行 使用以下代码从 Composer Airflow DAG 进行数据流作业。
我收到 2 种类型的错误消息,具体取决于代码。
请提出解决方法。
a) 错误 1:当服务帐户电子邮件被评论时 (#)
# "serviceAccountEmail": "service-7276363xxxxx@cloudcomposer-accounts.iam.gserviceaccount.com",
错误:
Error: Required 'compute.subnetworks.get' permission for 'projects/vpc-host/regions/us-central1/subnetworks/sbn-dataflow'
b) 错误 2:使用服务帐户电子邮件时
"serviceAccountEmail": "service-7276363xxxxx@cloudcomposer-accounts.iam.gserviceaccount.com",
错误:
Current user cannot act as service account service-7276363xxxxx@cloudcomposer-accounts.iam.gserviceaccount.com
代码:
import datetime
from airflow import models
from airflow.contrib.operators.dataflow_operator import DataflowTemplateOperator
from airflow.utils.dates import days_ago
bucket_path = models.Variable.get("bucket_path")
project_id = models.Variable.get("project_id")
gce_zone = models.Variable.get("gce_zone")
default_args = {
"owner": "Airflow",
"start_date": days_ago(1),
"depends_on_past": False,
"dataflow_default_options": {
"project": project_id,
"zone": gce_zone,
"serviceAccountEmail": "service-7276363xxxxx@cloudcomposer-accounts.iam.gserviceaccount.com",
"subnetwork": "https://www.googleapis.com/compute/v1/projects/vpc-host/regions/us-central1/subnetworks/sbn-dataflow",
"tempLocation": bucket_path + "/tmp/",
}
}
with models.DAG(
dag_id="composer_dataflow_dag",
default_args=default_args,
schedule_interval=datetime.timedelta(days=1)
) as dag:
dataflow_template_job = DataflowTemplateOperator(
task_id="dataflow_csv_to_bq",
template="gs://dataflow-templates/latest/GCS_Text_to_BigQuery",
parameters={
"javascriptTextTransformFunctionName": "transformCSVtoJSON",
"javascriptTextTransformGcsPath": bucket_path + "/SCORE_STG.js",
"JSONPath": bucket_path + "/SCORE_STG.json",
"inputFilePattern": bucket_path + "/stg_data.csv",
"outputTable": project_id + ":gcp_stage.SCORE_STG",
"bigQueryLoadingTemporaryDirectory": bucket_path + "/tmp/",
},
dag=dag,
)
您必须使用不同的服务 accounts.Remember 它必须能够访问资源。这应该可以解决这两个问题。
您可以按照 Role Assignment 中的说明创建一个服务帐户作为工作人员。即:一名工人和一名管理员。
除此之外,我没有发现任何异常情况。连参数都正确传递了。供其他用户参考:
我正在尝试 运行 使用以下代码从 Composer Airflow DAG 进行数据流作业。
我收到 2 种类型的错误消息,具体取决于代码。
请提出解决方法。
a) 错误 1:当服务帐户电子邮件被评论时 (#)
# "serviceAccountEmail": "service-7276363xxxxx@cloudcomposer-accounts.iam.gserviceaccount.com",
错误:
Error: Required 'compute.subnetworks.get' permission for 'projects/vpc-host/regions/us-central1/subnetworks/sbn-dataflow'
b) 错误 2:使用服务帐户电子邮件时
"serviceAccountEmail": "service-7276363xxxxx@cloudcomposer-accounts.iam.gserviceaccount.com",
错误:
Current user cannot act as service account service-7276363xxxxx@cloudcomposer-accounts.iam.gserviceaccount.com
代码:
import datetime
from airflow import models
from airflow.contrib.operators.dataflow_operator import DataflowTemplateOperator
from airflow.utils.dates import days_ago
bucket_path = models.Variable.get("bucket_path")
project_id = models.Variable.get("project_id")
gce_zone = models.Variable.get("gce_zone")
default_args = {
"owner": "Airflow",
"start_date": days_ago(1),
"depends_on_past": False,
"dataflow_default_options": {
"project": project_id,
"zone": gce_zone,
"serviceAccountEmail": "service-7276363xxxxx@cloudcomposer-accounts.iam.gserviceaccount.com",
"subnetwork": "https://www.googleapis.com/compute/v1/projects/vpc-host/regions/us-central1/subnetworks/sbn-dataflow",
"tempLocation": bucket_path + "/tmp/",
}
}
with models.DAG(
dag_id="composer_dataflow_dag",
default_args=default_args,
schedule_interval=datetime.timedelta(days=1)
) as dag:
dataflow_template_job = DataflowTemplateOperator(
task_id="dataflow_csv_to_bq",
template="gs://dataflow-templates/latest/GCS_Text_to_BigQuery",
parameters={
"javascriptTextTransformFunctionName": "transformCSVtoJSON",
"javascriptTextTransformGcsPath": bucket_path + "/SCORE_STG.js",
"JSONPath": bucket_path + "/SCORE_STG.json",
"inputFilePattern": bucket_path + "/stg_data.csv",
"outputTable": project_id + ":gcp_stage.SCORE_STG",
"bigQueryLoadingTemporaryDirectory": bucket_path + "/tmp/",
},
dag=dag,
)
您必须使用不同的服务 accounts.Remember 它必须能够访问资源。这应该可以解决这两个问题。
您可以按照 Role Assignment 中的说明创建一个服务帐户作为工作人员。即:一名工人和一名管理员。
除此之外,我没有发现任何异常情况。连参数都正确传递了。供其他用户参考: