GCP Composer（Airflow）连接GCP PubSub报错：conn_id `google_cloud_default`未定义，如何通过GCP服务账号获取连接ID认证？

嗨，我之前也踩过这个坑！这个报错的核心原因其实很直白：你用的PubSubPullSensor这类Google Cloud算子，默认会去找名为google_cloud_default的Airflow连接配置，但你的环境里还没创建这个连接，所以才会抛出异常。

下面我给你一步步讲怎么用GCP服务账号搞定这个连接配置，几种方式任你选：

第一步：准备好有权限的GCP服务账号 #

首先得在GCP控制台创建一个服务账号，给它分配Pub/Sub相关的必要权限：

• 至少要给它加上**Pub/Sub订阅者（roles/pubsub.subscriber）**角色，这样才能拉取并ACK订阅里的消息
• 如果你的DAG还要操作其他GCP资源，记得补充对应的权限

创建完成后，下载这个服务账号的JSON密钥文件，存好备用。

方式一：Airflow Web UI手动配置（最直观） #

因为你用的是GCP Composer，直接登录它对应的Airflow UI就行：

1. 点击顶部导航栏的「Admin」→「Connections」
2. 点击「+ Add a new record」新建连接
3. 填写关键信息：
   • Conn Id：必须填google_cloud_default（因为你的代码里没指定自定义conn_id，算子会默认用这个）
   • Conn Type：选择「Google Cloud Platform」
   • Keyfile JSON：把你刚才下载的服务账号JSON密钥内容粘贴进去
   • Project Id：填上你的GCP项目ID（也就是代码里的GCP_REPORTING_ID）
4. 其他字段保持默认，点击「Save」保存就行

方式二：用代码自动化创建连接（适合批量/部署场景） #

如果需要自动化配置，你可以写一段Python代码来创建连接，比如：

from airflow.models import Connection
from airflow.utils.db import create_session
import json

# 替换成你的服务账号JSON路径和项目ID
SERVICE_ACCOUNT_PATH = "/path/to/your/service-account-key.json"
GCP_PROJECT_ID = "your-project-id"

def create_gcp_default_connection():
    with open(SERVICE_ACCOUNT_PATH, "r") as f:
        keyfile_dict = json.load(f)
    
    with create_session() as session:
        # 检查连接是否已存在，避免重复创建
        existing_conn = session.query(Connection).filter(Connection.conn_id == "google_cloud_default").first()
        if not existing_conn:
            new_conn = Connection(
                conn_id="google_cloud_default",
                conn_type="google_cloud_platform",
                extra={
                    "extra__google_cloud_platform__project_id": GCP_PROJECT_ID,
                    "extra__google_cloud_platform__keyfile_dict": keyfile_dict,
                }
            )
            session.add(new_conn)
            session.commit()
            print("Successfully created google_cloud_default connection!")
        else:
            print("google_cloud_default connection already exists.")

# 可以在DAG初始化时调用，或者单独执行
create_gcp_default_connection()

注意：这种方式需要你的Airflow环境有创建连接的权限，在Composer里可能需要调整相关IAM权限。

方式三：自定义连接ID（不用默认名） #

如果你不想用google_cloud_default这个默认名，也可以在代码里手动指定自定义的conn_id：

1. 先按照上面的方式创建一个任意名字的GCP连接（比如my_pubsub_gcp_conn）
2. 修改你的DAG代码，加上conn_id参数：

subscribe_pubsub_task = PubSubPullSensor(
  task_id="subscribe_pubsub",
  project_id=GCP_REPORTING_ID,
  subscription=PUBSUB_SUBSCRIPTION,
  ack_messages=True,
  deferrable=True,
  conn_id="my_pubsub_gcp_conn"  # 替换成你自定义的连接ID
)

验证配置 #

配置完成后，重新触发DAG运行，那个conn_id未定义的报错应该就消失啦！

备注：内容来源于stack exchange，提问作者Alberto Sanmartin Martinez