为访问Azure SQL Database的GCP Cloud Function配置静态IP或窄IP段

一、创建GCP VPC及网络资源 #

• 创建自定义VPC网络
  在GCP控制台进入「VPC网络」页面，点击「创建VPC网络」：

  • 名称：比如cloud-function-vpc
  • 子网模式：自定义
  • 添加子网：选择与后续Cloud Function部署一致的区域（如us-central1），设置IP范围（如10.0.0.0/24）
  • 其余保持默认，完成创建

• 配置云NAT网关（实现固定出口IP）
  进入「VPC网络」→「云NAT」页面，点击「创建云NAT网关」：

  • 名称：比如cf-nat-gateway
  • VPC网络：选择刚才创建的cloud-function-vpc
  • 区域：与子网区域一致
  • 静态IP配置：
    • 若需单个静态IP：选择「指定静态IP地址」，点击「创建IP地址」生成新的静态公网IP
    • 若需窄IP段：选择「指定地址范围」，创建一个小型CIDR段（如34.123.45.0/28）并关联
  • 其余保持默认，完成创建

• 确认路由配置
  进入「VPC网络」→「路由表」，找到VPC对应的默认路由，确保下一跳为刚才创建的云NAT网关，保证出站流量通过NAT转发

二、部署Cloud Function到目标VPC #

部署Cloud Function时，在「运行时、构建、连接和安全设置」→「连接」部分：

• VPC连接：选择刚才创建的cloud-function-vpc及对应子网
• 出站流量：选择「通过VPC访问」
• 确保Cloud Function的部署区域与VPC子网区域完全一致，否则无法关联

三、获取对外出口IP/IP段 #

• 单个静态IP：进入「VPC网络」→「外部IP地址」，找到关联到云NAT网关的静态IP，记录公网IP地址
• IP段：进入「云NAT」网关详情页，查看关联的地址范围，记录CIDR格式的IP段

四、配置Azure SQL Database白名单 #

登录Azure门户，找到目标SQL Database：

• 进入「防火墙和虚拟网络」设置
• 关闭「允许Azure服务和资源访问此服务器」（提升安全性）
• 在「客户端IP地址」区域，点击「添加客户端IP」，输入刚才记录的静态IP或IP段，点击「保存」

五、Cloud Function代码实现要点 #

依赖配置（requirements.txt） #

sqlalchemy>=2.0.0
pyodbc>=4.0.39
google-cloud-bigquery>=3.11.0
pandas>=2.1.0
google-cloud-secret-manager>=2.16.0

核心代码示例 #

import sqlalchemy
from google.cloud import bigquery
import pandas as pd
from google.cloud import secretmanager

def get_secret(secret_id):
    # 从Secret Manager读取敏感配置
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/<你的GCP项目ID>/secrets/{secret_id}/versions/latest"
    response = client.access_secret_version(name=name)
    return response.payload.data.decode("UTF-8")

def azure_sql_to_bigquery(event, context):
    # 读取敏感配置
    azure_user = get_secret("azure-sql-username")
    azure_pwd = get_secret("azure-sql-password")
    azure_server = get_secret("azure-sql-server")
    azure_db = get_secret("azure-sql-database")
    
    # 构建Azure SQL连接字符串
    conn_str = f"mssql+pyodbc://{azure_user}:{azure_pwd}@{azure_server}.database.windows.net:1433/{azure_db}?driver=ODBC+Driver+17+for+SQL+Server"
    engine = sqlalchemy.create_engine(conn_str)
    
    # 读取Azure SQL数据
    query = "SELECT * FROM <目标表名>"
    with engine.connect() as conn:
        df = pd.read_sql(query, conn)
    
    # 写入BigQuery
    bq_client = bigquery.Client()
    table_id = "<你的GCP项目ID>.<BigQuery数据集ID>.<目标表名>"
    job = bq_client.load_table_from_dataframe(df, table_id)
    job.result()  # 等待数据加载完成
    
    return f"成功同步 {len(df)} 条数据到BigQuery"

注意事项 #

• 所有敏感信息（用户名、密码、服务器地址）必须存储在GCP Secret Manager中，禁止硬编码
• 为Cloud Function的服务账号分配Secret Manager的访问权限、BigQuery的写入权限

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