在Google Cloud Composer中为Airflow连接SQL数据库的问题求助

问题1：SQLite使用GCS路径失败 #

原因 #

SQLite是本地文件型数据库，仅支持访问本地文件系统路径，无法直接识别Google Cloud Storage（GCS）的gs://格式路径，因此测试连接和DAG运行时会提示无法找到数据库。

临时解决（仅适合测试场景） #

若坚持使用SQLite，需通过GCS FUSE将GCS Bucket挂载到Composer Worker的本地目录：

• 在Composer环境中启用GCS FUSE，将目标Bucket挂载至Worker本地路径（如/mnt/gcs-sqlite）
• Airflow WebUI中SQLite连接的Host字段填写挂载后的本地文件路径（如/mnt/gcs-sqlite/your-db-file.db）

生产环境不推荐SQLite，分布式Worker并发访问会导致数据一致性问题，建议改用Cloud SQL。

问题2：Google Cloud SQL连接配置 #

前置条件 #

• Composer环境已安装apache-airflow-providers-google插件（默认已包含）
• Cloud SQL实例已创建并导入数据库，且允许Composer Worker的VPC/IP段访问

Airflow WebUI连接配置（以PostgreSQL为例，MySQL配置逻辑一致） #

1. 新建连接，选择Google Cloud SQL (PostgreSQL)（MySQL选对应类型）
2. 填写核心参数：
   • Conn ID：自定义ID（如episodes_cloudsql）
   • 项目ID：你的GCP项目ID
   • 实例ID：Cloud SQL完整ID（格式：项目ID:区域:实例名）
   • 数据库：导入的目标数据库名称
   • 登录/密码：Cloud SQL数据库账号凭证
   • （可选）若启用IAM数据库认证，勾选「使用IAM认证」，无需填写密码

DAG代码适配 #

替换原SQLite操作器为Cloud SQL对应类型：

from airflow.providers.postgres.operators.postgres import PostgresOperator
import requests
import xmltodict
from airflow.decorators import task

PODCAST_URL = "你的播客RSS地址"

def podcast_summary():
    create_table = PostgresOperator(
        task_id='create_table_postgres',
        sql=r"""
        CREATE TABLE IF NOT EXISTS episodes (
            link TEXT PRIMARY KEY,
            title TEXT,
            filename TEXT,
            published TEXT,
            description TEXT,
            transcript TEXT
        );
        """,
        postgres_conn_id="episodes_cloudsql"
    )

    @task()
    def get_episodes():
        data = requests.get(PODCAST_URL)
        feed = xmltodict.parse(data.text)
        episodes = feed["rss"]["channel"]["item"]
        print(f"发现 {len(episodes)} 个剧集。")
        return episodes

    podcast_episodes = get_episodes()
    create_table.set_downstream(podcast_episodes)

summary = podcast_summary()

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