GCP新手求教：如何通过Cloud Composer在多台Compute Engine运行DAG中的Python脚本

• 核心逻辑：将Cloud Composer DAG目录（对应GCS存储桶）中的Python脚本同步到目标Compute Engine实例，再通过SSH执行，避免在每台实例上维护本地脚本。

• 分步实现：

  1. 脚本同步
     Cloud Composer的DAG目录默认映射到gs://<你的Composer存储桶名>/dags/，在DAG中可以用GSUtilOperator或BashOperator调用gsutil cp命令，把脚本复制到CE实例的临时目录（比如/tmp/）。要给CE实例的服务账号添加roles/storage.objectViewer权限，确保能访问这个GCS桶。

  2. SSH执行脚本
     脚本同步完成后，用SSHOperator连接CE实例执行脚本。示例代码：

     SSHOperator(
         task_id="run_python_script",
         ssh_conn_id="ce_ssh_connection",
         command="python3 /tmp/your_script.py"
     )
     

     提前在Airflow连接中配置好CE实例的SSH信息，推荐用服务账号密钥实现免密登录。

  3. 完整工作流编排
     把各步骤按顺序串起来：

     • 存储桶触发：用GoogleCloudStorageObjectSensor监听指定桶的文件上传事件
     • 启动CE实例：ComputeEngineStartInstanceOperator
     • 同步脚本到CE实例：执行gsutil cp的操作符
     • 运行脚本：SSHOperator
     • 关闭CE实例：ComputeEngineStopInstanceOperator

  4. 多实例批量处理优化
     针对30多台实例，用Airflow的DynamicTaskMapping动态生成任务，不用重复编写代码。示例：

     target_instances = ["instance-01", "instance-02", ..., "instance-30"]
     
     # 批量启动实例
     start_instances = ComputeEngineStartInstanceOperator.partial(
         task_id="start_ce_instance"
     ).expand(instance_name=target_instances)
     
     # 同步脚本、执行脚本、关闭实例的任务都可以用同样方式批量生成
     

• 权限要点

  • 给Cloud Composer的服务账号添加roles/compute.instanceAdmin.v1权限，允许操作CE实例的启停
  • CE实例的服务账号需具备GCS DAG桶的读取权限
  • 若用SSH免密登录，将Composer环境的公钥添加到CE实例的~/.ssh/authorized_keys中

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