如何传入OAUTH ACCESS TOKEN凭证运行GCP Dataflow管道

原因说明 #

你遇到的invalid_grant+invalid_rapt报错是因为PipelineOptions默认没有读取你设置的GOOGLE_OAUTH_ACCESS_TOKEN环境变量，仍然走了默认的凭证刷新流程，当前使用的凭证触发了重鉴权要求无法正常刷新。

配置步骤 #

方法1：直接通过GoogleCloudOptions传入access token（适用于Apache Beam 2.20+版本） #

1. 导入os模块读取环境变量
2. 在你设置GoogleCloudOptions的代码段中新增access_token配置即可，修改后的代码如下：

import os
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions, StandardOptions, SetupOptions

def run(argv=None):
    pipeline_options = PipelineOptions(flags=argv)
    google_cloud_options = pipeline_options.view_as(GoogleCloudOptions)
    google_cloud_options.project = 'prjoject name'
    google_cloud_options.region = 'region'
    google_cloud_options.job_name = 'jobname'
    google_cloud_options.staging_location = 'gs://staging-location' 
    google_cloud_options.temp_location = 'gs://temp-location'
    # 新增：传入环境变量中的OAUTH ACCESS TOKEN
    google_cloud_options.access_token = os.getenv('GOOGLE_OAUTH_ACCESS_TOKEN')
    
    pipeline_options.view_as(StandardOptions).runner = 'DataflowRunner'
    pipeline_options.view_as(SetupOptions).save_main_session = True
    pipeline_options.view_as(SetupOptions).setup_file = "./setup.py"
    user_options = pipeline_options.view_as(DataInput)

    p = beam.Pipeline(options=pipeline_options)
    
    # 后续业务逻辑不变
    input_meta = str(user_options.input)+" "+project+ " "+bucket_name+ " "+ model_path+ " "+destination_name
    header = (p
          | "CREATE BEAM" >> beam.Create([input_meta]) 
          | "Inside read header" >> beam.Map(read_header_from_filename))

    result = p.run()
    result.wait_until_finish()

方法2：自定义凭证对象（适用于旧版本Beam） #

如果你的Beam版本低于2.20，可以手动构造凭证对象传入配置：

import os
from google.oauth2.credentials import Credentials

# 在初始化PipelineOptions后添加如下代码
credentials = Credentials(os.getenv('GOOGLE_OAUTH_ACCESS_TOKEN'))
google_cloud_options.gcp_credentials = credentials

注意事项 #

• 确认你使用的access token有效，且具备Dataflow任务创建、GCS读写等必要权限，token过期时间需要覆盖管道的完整运行周期：静态access token不会自动刷新，若需要运行长时间任务，更推荐使用服务账号密钥文件，设置GOOGLE_APPLICATION_CREDENTIALS环境变量指向密钥文件路径即可自动处理凭证刷新。
• invalid_rapt报错多出现于使用个人用户账号生成的token场景，这类token可能触发二次验证限制，切换为服务账号生成的access token即可避免该问题。

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