Google Cloud Composer环境下Airflow DAG找不到connection ID报错咨询

第一步：确认Composer环境级别的连接配置差异 #

• 先校验当前故障环境的Airflow连接配置参数是否匹配正常运行的项目配置：
  • 确认故障环境的[redacted-name]连接的连接类型是否为Google Cloud，注意不要误选成Google Cloud Platform这类旧版本别名类型
  • 检查连接的Keyfile Path/Keyfile JSON字段是否正确填入了完整的服务账号密钥内容，注意如果用Keyfile JSON字段，不要在前后添加任何换行符、引号包裹，直接粘贴JSON原始内容即可
• 对比两个项目的Airflow版本差异：如果两个Composer环境对应的Airflow大版本不同（比如一个是Airflow 2.x一个是1.10.x），Google provider包版本差异会导致连接读取逻辑不同，可运行命令pip show apache-airflow-providers-google分别查询两个环境的包版本，确认是否一致

第二步：排查DAG代码的隐式配置 #

• 检查故障DAG是否在代码中硬写了连接前缀/后缀，比如部分开发者会在代码中通过os.environ读取环境变量拼接连接ID，导致实际读取的连接名和Airflow后台配置的不一致
• 确认DAG所用的Operator是否是对应版本的官方Operator：如果用了自定义Operator或者过时的BigQueryOperator而非BigQueryInsertJobOperator，会存在连接读取逻辑的兼容性问题
• 可以在故障DAG中添加临时调试任务，打印连接ID的实际读取值：

from airflow.hooks.base import BaseHook
from airflow.operators.python import PythonOperator

def debug_connection(**kwargs):
    conn_id = "[redacted-name]" # 替换成实际的连接ID
    print(f"Trying to read connection: {conn_id}")
    try:
        conn = BaseHook.get_connection(conn_id)
        print(f"Connection found: {conn.conn_type}")
    except Exception as e:
        print(f"Connection read error: {e}")

debug_task = PythonOperator(
    task_id="debug_gcp_connection",
    python_callable=debug_connection,
    dag=dag
)

运行该任务后查看日志，确认连接是否能被正常读取，如果该调试任务也报找不到连接，说明是环境侧的连接配置问题，否则是DAG任务的连接传参问题。

第三步：排查Composer环境的权限配置 #

• 确认当前Composer环境的工作负载身份配置是否正确：如果开启了工作负载身份，需要确认Airflow worker的Kubernetes服务账号有权限读取Secret Manager中存储的连接凭证（如果你的Composer配置了将连接存储在Secret Manager中）
• 检查Airflow的core.sql_alchemy_conn配置是否正常，是否存在元数据库读取异常导致连接信息加载失败的情况

实际案例中同类报错的常见根因是两个项目的apache-airflow-providers-google版本差异，旧版本不支持在连接中直接粘贴Keyfile JSON内容，只能用Keyfile Path路径读取，升级provider包后即可解决。

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