如何在现有Google Cloud VM上运行Apache Beam Python管道以访问Cloud SQL

首先明确：Dataflow托管服务不支持指定现有VM实例运行作业，Worker节点由服务自动调度管理，没法直接绑定自定义VM。你需要通过配置Worker的网络权限或使用代理的方式，让自动创建的Worker能访问Cloud SQL，具体方案如下：

方案1：配置Dataflow Worker的网络与权限 #

通过启动参数让Worker在有权限的网络环境中运行，并赋予对应的访问角色：

• 指定子网：启动作业时添加--subnetwork=regions/<区域>/subnetworks/<子网名称>，确保该子网与Cloud SQL实例在同一VPC（或已通过VPC peering连通）
• 指定服务账号：添加--service_account_email=<有权限的服务账号邮箱>，该账号需要拥有roles/cloudsql.client角色（允许访问Cloud SQL实例）
• 启用Cloud SQL Private IP：如果Cloud SQL实例配置了Private IP，Worker在同一VPC下无需公网即可直接访问，避免公网权限问题

示例启动命令片段：

python your_pipeline.py \
  --runner=DataflowRunner \
  --project=your-project-id \
  --region=us-central1 \
  --subnetwork=regions/us-central1/subnetworks/your-subnet \
  --service_account_email=dataflow-worker-sa@your-project-id.iam.gserviceaccount.com \
  --temp_location=gs://your-bucket/temp

方案2：在Dataflow Worker中运行Cloud SQL Auth Proxy #

通过在Beam管道的Worker节点上启动Auth Proxy，实现无直接网络权限下的Cloud SQL访问：

• 在自定义DoFn的setup()方法中启动Auth Proxy进程，teardown()方法中关闭进程
• 需要确保Worker节点有下载Auth Proxy的权限（默认能访问公网即可）

示例代码片段：

import subprocess
import apache_beam as beam

class CloudSQLIngestDoFn(beam.DoFn):
    def setup(self):
        # 下载并启动Cloud SQL Auth Proxy
        subprocess.run(["wget", "https://dl.google.com/cloudsql/cloud_sql_proxy.linux.amd64", "-O", "/tmp/cloud_sql_proxy"])
        subprocess.run(["chmod", "+x", "/tmp/cloud_sql_proxy"])
        self.proxy_process = subprocess.Popen(
            ["/tmp/cloud_sql_proxy", "-instances=your-project-id:us-central1:your-sql-instance=tcp:5432"],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

    def process(self, element):
        # 这里用localhost:5432连接Cloud SQL
        # 执行数据摄入逻辑
        pass

    def teardown(self):
        # 关闭Auth Proxy进程
        self.proxy_process.terminate()

方案3：预导出Cloud SQL数据到GCS（替代方案） #

如果上述方案都无法快速落地，可以先将Cloud SQL数据导出到GCS（通过Cloud SQL导出功能或自定义脚本），再让Dataflow直接读取GCS中的数据，绕过VM访问Cloud SQL的权限问题。

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