如何通过Cloud Composer DAG使用服务账号执行API调用并运行Dataflow任务

问题1：是否需要创建新服务账号？ #

不需要额外创建新的服务账号，直接使用Composer环境自带的默认服务账号即可（创建Composer环境时自动生成，格式通常为service-<项目编号>@gcp-sa-cloudcomposer.iam.gserviceaccount.com）。你只需要给这个账号添加以下必要的IAM角色：

• roles/dataflow.developer：允许创建、启动和管理Dataflow作业
• BigQuery相关角色：如果仅读取数据，添加roles/bigquery.dataViewer + roles/bigquery.jobUser；如果需要写入BigQuery，补充roles/bigquery.dataEditor
• GCS相关角色：根据实际需求添加roles/storage.objectAdmin（全权限），或更细粒度的roles/storage.objectCreator + roles/storage.objectViewer（仅读写对象）

注意：Dataflow工作者的服务账号权限保持不变即可，Composer的服务账号只需要具备触发Dataflow作业的权限，不需要接管工作者的执行权限。

问题2：无密钥文件的最佳认证方式？ #

使用GCP官方推荐的默认应用凭据（ADC），完全不需要在代码中传递JSON密钥文件，具体操作如下：

1. 确保Composer环境的服务账号已配置好上述所有必要权限
2. 在PythonOperator的代码中，调用Dataflow相关API（无论是用Apache Beam构建Pipeline，还是用google-cloud-dataflow客户端）时，不要显式指定凭据参数，ADC会自动获取Composer环境服务账号的身份
3. 示例（Apache Beam Pipeline）：

from apache_beam import Pipeline, runners

pipeline_options = {
    'project': '你的GCP项目ID',
    'region': '你的区域',
    'temp_location': 'gs://你的GCS桶/temp/',
    'runner': 'DataflowRunner'
}

with Pipeline(options=runners.DataflowRunner().get_default_options(**pipeline_options)) as p:
    # 你的Dataflow处理逻辑
    pass

之前的认证错误，大概率是因为Composer服务账号权限不足，或者代码中手动指定了其他凭据导致冲突，检查代码并移除手动指定凭据的逻辑即可。

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