Google Cloud Composer中cloud-sql-python-connector模块导入失败求助

问题背景 #

通过Google Cloud Composer调度Dataflow管道，从Cloud SQL SQL Server实例读取数据写入BigQuery，使用google.cloud.sql.connector[pytds]建立连接。手动在Cloud Shell运行Dataflow任务成功，但通过Cloud Composer触发时出现NameError: name 'Connector' is not defined。

已确认配置：

• 启用save_main_session选项
• py_requirements已包含相关依赖：

  py_requirements=['apache-beam[gcp]==2.41.0','cloud-sql-python-connector[pytds]==0.6.1','pyodbc==4.0.34','SQLAlchemy==1.4.41','pymssql==2.2.5','sqlalchemy-pytds==0.3.4','pylint==2.15.4']
  

• Airflow日志显示cloud-sql-python-connector[pytds]==0.6.1已完成安装

解决方案 #

1. 显式完整导入Connector

Cloud Composer与本地/Cloud Shell的导入上下文存在差异，必须在Dataflow代码的顶层位置显式指定完整导入路径：

# 正确导入方式
from google.cloud.sql.connector import Connector

避免仅写import Connector或在函数内部导入，确保模块在会话初始化时就被加载。

2. 确保Dataflow Worker节点获取依赖

Composer环境安装的依赖仅作用于Airflow调度层，Dataflow的Worker节点是独立环境，需确保依赖传递给Worker：

• 若使用DataflowPythonOperator，确认py_requirements参数正确传递给任务；
• 若使用setup.py管理依赖，需在install_requires中明确添加：

  install_requires=[
      'apache-beam[gcp]==2.41.0',
      'cloud-sql-python-connector[pytds]==0.6.1',
      # 其他必要依赖
  ]
  

3. 验证save_main_session的生效范围

启用save_main_session=True后，需确保Connector的导入在主脚本顶层，而非函数内部。示例正确结构：

# 顶层导入，确保会话初始化时加载
from google.cloud.sql.connector import Connector
import apache_beam as beam

def run_pipeline():
    connector = Connector()
    # 管道逻辑...

if __name__ == '__main__':
    run_pipeline()

4. 检查Python版本兼容性

cloud-sql-python-connector[pytds]==0.6.1仅支持Python 3.7+，若Composer环境使用Python 3.6或更低版本，会导致模块无法正常导入。可通过Composer控制台查看环境Python版本，必要时升级环境。

5. 清理依赖缓存重新安装

依赖缓存可能导致异常，可在py_requirements中添加--no-cache-dir强制重新安装：

py_requirements=[
    'apache-beam[gcp]==2.41.0',
    'cloud-sql-python-connector[pytds]==0.6.1 --no-cache-dir',
    # 其他依赖
]

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