Apache Airflow连接GCP Cloud SQL MySQL超时问题求助

核心问题分析 #

本地运行正常但Airflow中连接超时，核心原因是Airflow的运行环境与本地的网络出口/权限配置不同：

• 本地机器的公网IP已被加入Cloud SQL白名单，而Airflow所在环境的出口IP不在白名单内
• 若使用托管式Airflow（如Cloud Composer），其工作节点没有固定公网IP，直接加白名单无效

可行解决方案 #

方案1：使用Cloud SQL Auth Proxy（推荐，安全且无需白名单） #

Auth Proxy通过GCP服务账号权限建立加密连接，无需暴露数据库公网IP或配置白名单，适配所有Airflow部署场景：

自建Airflow环境实现方式

在Python任务中启动Proxy进程后再连接数据库：

import mysql.connector
from google.cloud import storage
import pandas as pd
import subprocess
import time

def import_csv_to_sql():
    # 启动Cloud SQL Auth Proxy（需提前在Airflow节点安装proxy）
    proxy_process = subprocess.Popen([
        '/usr/local/bin/cloud_sql_proxy',
        f'-instances=你的项目ID:区域:实例名=tcp:3306',
        '-credential_file=/opt/airflow/secrets/service-account-key.json'
    ])
    time.sleep(3)  # 等待代理初始化

    try:
        # 从GCS读取CSV
        storage_client = storage.Client()
        bucket = storage_client.get_bucket('你的存储桶名')
        df = pd.read_csv(bucket.blob('data.csv').open())

        # 通过本地代理连接MySQL（host为127.0.0.1）
        conn = mysql.connector.connect(
            host='127.0.0.1',
            user='数据库用户名',
            password='数据库密码',
            database='目标数据库名'
        )
        cursor = conn.cursor()
        # 执行数据导入逻辑
        for _, row in df.iterrows():
            cursor.execute(
                "INSERT INTO 目标表 (列1, 列2) VALUES (%s, %s)",
                (row['列1'], row['列2'])
            )
        conn.commit()
    finally:
        # 确保代理进程关闭
        proxy_process.terminate()
        proxy_process.wait()

Cloud Composer（托管Airflow）实现方式

Composer已预装Auth Proxy，且默认服务账号可配置Cloud SQL权限：

1. 给Composer的服务账号添加Cloud SQL Client角色
2. 使用Airflow官方CloudSqlInstanceImportOperator直接完成导入，无需手动处理连接：

from airflow.providers.google.cloud.operators.cloud_sql import CloudSqlInstanceImportOperator

import_task = CloudSqlInstanceImportOperator(
    task_id='import_gcs_csv_to_cloudsql',
    project_id='你的项目ID',
    instance='你的Cloud SQL实例名',
    body={
        "importContext": {
            "fileType": "CSV",
            "uri": "gs://你的存储桶名/data.csv",
            "database": "目标数据库名",
            "csvImportOptions": {
                "table": "目标表名",
                "columns": ["列1", "列2"]  # 可选：指定CSV列对应表列
            }
        }
    }
)

方案2：配置白名单（仅适用于有固定出口IP的Airflow环境） #

如果坚持用公网IP连接，需找到Airflow的实际出口IP并加入Cloud SQL白名单：

• 自建GCE实例上的Airflow：直接取GCE实例的公网IP
• GKE上的Airflow：若使用Cloud NAT，取NAT网关的公网IP；否则取节点的公网IP
• Cloud Composer：需配置Cloud NAT，取NAT网关的公网IP（Composer默认无固定出口IP，必须配置NAT才能用白名单方式）

排查验证步骤 #

1. 测试网络连通性：在Airflow任务中添加端口测试命令，确认能否访问Cloud SQL的3306端口：

nc -zv 你的Cloud SQL公网IP 3306

2. 核对配置信息：确认Airflow中使用的数据库IP、用户名、密码与本地一致，避免环境变量配置错误
3. 检查权限：确保Airflow使用的服务账号（或节点）有Cloud SQL连接权限（若用Auth Proxy或VPC连接）

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