Google Dataflow创建GCS模板缺失storage.objects.create权限

算力阿峰

2026-8-28

问题背景

在Google Cloud Dataflow上构建Apache Beam管道，创建模板上传至GCS时触发权限报错：Cloud Dataflow无权写入模板目录，缺失storage.objects.create访问权限。
Dataflow权限报错截图
已完成的排查操作：

为对应服务账号授予了Storage Admin和Viewer权限
所有相关API均已启用
尝试移除服务账号的Storage Admin角色后重新授权，问题仍然存在

使用的完整管道代码如下：

from __future__ import annotations

import json
from typing import TYPE_CHECKING
import os

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.options.pipeline_options import SetupOptions

if TYPE_CHECKING:
    from apache_beam.options.pipeline_options import _BeamArgumentParser

os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = <apikey.json>

class ArgumentParser(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser: _BeamArgumentParser) -> None:
        parser.add_value_provider_argument(
            '--input_file',
            help='Path to the file to ingest data from',
            default='gs://dataflow_marketing_datamart/json_to_bq_test/input_data.jsonl',
            type=str
        )
        parser.add_value_provider_argument(
            '--bq_table',
            help='Output BigQuery table in the form of <PROJECT>:<DATASET>.<TABLE>',
            default='marketing-datamart:dataflow_testing.custom_template_test',
            type=str
        )
        parser.add_value_provider_argument(
            '--bq_schema',
            help='JSON string of the BigQuery table',
            default="""
            {
                "fields": [
                    {
                    "description": "Name of the fruit",
                    "name": "fruit",
                    "type": "STRING",
                    "mode": "REQUIRED"
                    },
                    {
                    "description": "Quantity of the fruit",
                    "name": "quantity",
                    "type": "INTEGER",
                    "mode": "NULLABLE"
                    },
                    {
                    "description": "Color of the fruit",
                    "name": "color",
                    "type": "STRING",
                    "mode": "NULLABLE"
                    }
                ]
            }""",
            type=str
        )



class FormatInputText(beam.DoFn):
    """beam.io.WriteToBigQuery expects a list of one dictionary, but the raw output from 
    beam.io.ReadFromText is string. This converts the string to the required format"""
    def process(self, line):
        return [json.loads(line)]


def main(argv=None, save_main_session=True):
    """Main entry point"""

    pipeline_args = []
    pipeline_args.extend([
        '--runner=DataflowRunner',
        '--project=$PROJECTt',
        '--region=asia-southeast1',
        '--staging_location=$BUCKET/staging',
        '--temp_location=$BUCKET/temp',
        '--job_name=custom-job-test',
        '--template_location=$BUCKET/template/trial3_template'
    ])

    # We use the save_main_session option because one or more DoFn's in this
    # workflow rely on global context (e.g., a module imported at module level).
    pipeline_options = PipelineOptions(pipeline_args)
    pipeline_options.view_as(SetupOptions).save_main_session = save_main_session

    template_options = pipeline_options.view_as(ArgumentParser)
    template_options.view_as(SetupOptions).save_main_session = save_main_session

    with beam.Pipeline(options=pipeline_options) as p:
        
        input_lines = (p 
                    | "Read input schema" >> beam.io.ReadFromText(template_options.input_file)
                    | "Format lines" >> beam.ParDo(FormatInputText()))

        bq_write = input_lines | "Write to BigQuery" >> beam.io.WriteToBigQuery(
            table=lambda x: f"{template_options.bq_table.get()}",
            schema=lambda x: json.loads(template_options.bq_schema.get()),
            write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND,
            create_disposition=beam.io.BigQueryDisposition.CREATE_IF_NEEDED
        )


if __name__ == '__main__':
  main()

解决方案

按以下顺序排查修复：

核对授权的服务账号是否正确：上传Dataflow模板的操作由本地执行脚本的身份（即GOOGLE_APPLICATION_CREDENTIALS指向的服务账号）完成，不是Dataflow运行时使用的服务代理账号。多数权限配置错误都是把角色加给了Dataflow服务代理，而非实际执行上传操作的账号。
检查GCS桶权限配置：不要只依赖项目级的角色继承，直接到模板、staging、temp路径所在的GCS桶权限页，给执行上传操作的服务账号直接授予storage.objects.create、storage.objects.list、storage.objects.get权限，同时确认桶级没有配置针对该账号的拒绝策略（Deny Policy）。权限配置完成后，可先用该服务账号的密钥执行gsutil cp 本地测试文件 gs://<你的桶路径>/test.txt验证上传权限，该命令能跑通再继续后续操作。
修复代码中的配置错误：
1. pipeline_args里的--project=$PROJECTt多写了一个尾缀t，会导致项目ID解析错误，请求落到错误的项目上下文触发权限校验失败
2. 参数里的$PROJECT、$BUCKET是Shell变量语法，Python不会自动替换这类占位符，需要通过os.getenv()读取环境变量赋值，或者直接替换为实际的项目ID、GCS桶路径
3. os.environ["GOOGLE_APPLICATION_CREDENTIALS"]对应的<apikey.json>是占位符，需要替换为服务账号密钥文件的本地绝对路径，否则SDK会读取系统默认身份，和你预期授权的账号不匹配。

注意：IAM权限变更后最长需要2分钟左右生效，不要刚改完权限立刻跑命令，等待1-2分钟再重试。

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