GCP Airflow Composer中BigQuery to_dataframe()报权限错误排查

问题 #

在GCP Airflow Composer中使用BigQuery Client将查询结果转为DataFrame时遇到权限异常：

• 本地无Airflow环境下代码运行正常
• Airflow中client.query(query)可正常执行，但调用client.query(query).to_dataframe()时抛出bigquery.readsessions.create权限拒绝错误

本地正常但Airflow中报错的代码 #

from google.cloud import bigquery

client = bigquery.Client(project="GCP PROJECT NAME")

# 获取所有访谈ID的品牌形象得分
query="""SELECT * from Table LIMIT 10"""

df_bi=client.query(query).to_dataframe()

print("Success")

实际报错信息 #

[2023-04-20, 12:47:25 UTC] {taskinstance.py:1853} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/opt/python3.8/lib/python3.8/site-packages/google/api_core/grpc_helpers.py", line 57, in error_remapped_callable
    return callable_(*args, **kwargs)
  File "/opt/python3.8/lib/python3.8/site-packages/grpc/_channel.py", line 946, in __call__
    return _end_unary_response_blocking(state, call, False, None)
  File "/opt/python3.8/lib/python3.8/site-packages/grpc/_channel.py", line 849, in _end_unary_response_blocking
    raise _InactiveRpcError(state)
grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
    status = StatusCode.PERMISSION_DENIED
    details = "request failed: the user does not have 'bigquery.readsessions.create' permission for 'projects/projectname'"

解决方案 #

原因分析 #

client.query(query)仅完成查询提交和元数据获取，无需创建读取会话；而to_dataframe()方法在处理查询结果时，会通过创建**读取会话（Read Session）**实现高效流式数据传输，这一操作需要bigquery.readsessions.create权限。本地环境使用的账号权限通常更完整，而Airflow Composer对应的服务账号缺少该权限。

解决步骤 #

1. 为Airflow服务账号补充权限
   • 找到Airflow Composer对应的服务账号（格式通常为composer-{环境名}@{项目ID}.iam.gserviceaccount.com）
   • 在GCP IAM控制台中，为该账号添加BigQuery Session User角色（该角色包含bigquery.readsessions.create权限），或直接添加bigquery.readsessions.create单独权限
2. 验证权限生效
   • 权限添加后等待数分钟（GCP权限同步存在延迟），重新运行Airflow任务
3. 替代方案（无法添加权限时）
   • 绕过to_dataframe()的读取会话逻辑，手动转换结果：

     query_job = client.query(query)
     results = query_job.result()
     df_bi = results.to_dataframe()
     # 或手动构造DataFrame：import pandas as pd; df_bi = pd.DataFrame([dict(row) for row in results])
     

     注：该方式在处理超大结果集时效率低于读取会话，但可绕过权限限制

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