Cloud Composer同一环境下能否使用不同服务账号运行DAG？

同一个Cloud Composer环境下完全可以实现部分DAG/KubernetesPodOperator任务使用非默认的自定义服务账号运行，以下是两种可落地的实现方式：

方案1：为KubernetesPodOperator单独挂载自定义服务账号密钥 #

该方案适用于各类需要指定服务账号的场景，配置方式如下：

• 提前在GCP IAM中创建自定义服务账号，配置完成所需权限后导出JSON格式的密钥文件，上传到Composer环境的/home/airflow/gcs/data共享目录
• 在KubernetesPodOperator参数中通过挂载卷的方式将密钥文件传入任务Pod，同时配置环境变量GOOGLE_APPLICATION_CREDENTIALS指向密钥路径即可生效
  代码示例：

from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from airflow.kubernetes.volume import Volume
from airflow.kubernetes.volume_mount import VolumeMount

# 定义共享存储挂载配置
sa_key_volume = Volume(
    name="sa-key-volume",
    configs={
        "persistentVolumeClaim": {
            "claimName": "composer-shared-pvc"
        }
    }
)
sa_key_volume_mount = VolumeMount(
    name="sa-key-volume",
    mount_path="/var/secrets/",
    sub_path="data/",
    read_only=True
)

custom_sa_task = KubernetesPodOperator(
    task_id="run-with-custom-sa",
    name="custom-sa-pod",
    image="gcr.io/google.com/cloudsdktool/cloud-sdk:slim",
    cmds=["gcloud", "auth", "list"], # 可通过该命令验证生效的服务账号
    volumes=[sa_key_volume],
    volume_mounts=[sa_key_volume_mount],
    env_vars={
        "GOOGLE_APPLICATION_CREDENTIALS": "/var/secrets/你的自定义服务账号密钥文件名.json"
    },
    is_delete_operator_pod=True,
    namespace="composer-user-workloads",
    get_logs=True
)

方案2：使用服务账号模拟（Impersonation） #

该方案不需要导出和管理密钥，安全性更高，仅需要提前给Composer默认环境服务账号授予目标自定义服务账号的roles/iam.serviceAccountTokenCreator权限即可：

• 如果使用GCP官方提供的Airflow Operator，直接传入impersonation_chain参数即可生效，示例：

from airflow.providers.google.cloud.operators.bigquery import BigQueryExecuteQueryOperator

test_query = BigQueryExecuteQueryOperator(
    task_id="test-impersonate-sa",
    sql="SELECT CURRENT_SERVICE_ACCOUNT()",
    use_legacy_sql=False,
    location="US",
    impersonation_chain="自定义服务账号ID@项目ID.iam.gserviceaccount.com"
)

• 如果是通用KubernetesPodOperator任务，直接在执行命令前开启服务账号模拟即可，示例：

custom_sa_impersonate_task = KubernetesPodOperator(
    task_id="run-with-impersonated-sa",
    name="impersonate-sa-pod",
    image="gcr.io/google.com/cloudsdktool/cloud-sdk:slim",
    cmds=[
        "bash",
        "-c",
        """
        gcloud config set auth/impersonate_service_account 自定义服务账号ID@项目ID.iam.gserviceaccount.com
        # 后续执行的所有GCP相关请求都会自动使用模拟的服务账号
        gcloud auth list
        """
    ],
    is_delete_operator_pod=True,
    namespace="composer-user-workloads",
    get_logs=True
)

注意：使用该方案前需确认Composer默认服务账号已获得目标自定义服务账号的服务账号令牌创建者权限，否则会触发权限报错。

内容的提问来源于stack exchange，提问作者Hutch