咨询Composer Python SDK官方文档及V2版本操作示例

前置依赖与认证 #

• 安装官方Python客户端库：

  pip install google-cloud-composer
  

• 本地开发认证：执行gcloud auth application-default login完成身份验证；生产环境建议使用服务账号密钥文件，通过设置环境变量GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account-key.json"实现认证。

1. 列出所有Composer V2环境 #

from google.cloud import composer_v1

def list_composer_environments(project_id, region):
    client = composer_v1.EnvironmentsClient()
    parent = client.common_location_path(project_id, region)
    
    # 过滤仅显示V2环境
    environments = client.list_environments(parent=parent)
    
    for env in environments:
        if env.config.environment_type == "COMPOSER_V2":
            print(f"环境名称: {env.name}")
            print(f"环境状态: {env.state}")
            print(f"Airflow版本: {env.config.software_configuration.airflow_version}")
            print("---")

# 替换为你的项目ID和区域
list_composer_environments("your-project-id", "us-central1")

2. 创建Composer V2环境 #

from google.cloud import composer_v1
from google.cloud.composer_v1.types import Environment, EnvironmentConfig, SoftwareConfiguration, NodeConfig

def create_composer_v2_environment(project_id, region, env_name):
    client = composer_v1.EnvironmentsClient()
    parent = client.common_location_path(project_id, region)
    
    # 配置V2环境核心参数
    software_config = SoftwareConfiguration(
        airflow_version="2.6.3",  # 替换为支持的Airflow版本
        python_version="3.10",
        environment_variables={
            "AIRFLOW__CORE__LOAD_EXAMPLES": "False"
        }
    )
    
    node_config = NodeConfig(
        machine_type="n2-standard-2",
        disk_size_gb=30,
        network="default",
        subnetwork="default"
    )
    
    env_config = EnvironmentConfig(
        environment_type="COMPOSER_V2",
        software_configuration=software_config,
        node_config=node_config,
        location=region
    )
    
    environment = Environment(config=env_config)
    
    # 发起创建请求
    operation = client.create_environment(
        request={"parent": parent, "environment_id": env_name, "environment": environment}
    )
    
    print(f"创建环境操作已启动，操作ID: {operation.operation.name}")
    operation.result()  # 等待创建完成
    print("环境创建成功")

# 替换为你的项目ID、区域和环境名称
create_composer_v2_environment("your-project-id", "us-central1", "my-composer-v2-env")

3. 更新Composer V2环境（示例：修改Airflow环境变量） #

from google.cloud import composer_v1
from google.protobuf.field_mask_pb2 import FieldMask

def update_composer_v2_environment(project_id, region, env_name):
    client = composer_v1.EnvironmentsClient()
    environment_name = client.environment_path(project_id, region, env_name)
    
    # 构建更新后的配置
    updated_env = client.get_environment(name=environment_name)
    updated_env.config.software_configuration.environment_variables["AIRFLOW__CORE__DAGBAG_IMPORT_TIMEOUT"] = "120"
    
    # 指定更新的字段路径
    update_mask = FieldMask(paths=["config.software_configuration.environment_variables"])
    
    # 发起更新请求
    operation = client.update_environment(
        request={"environment": updated_env, "update_mask": update_mask}
    )
    
    print(f"更新环境操作已启动，操作ID: {operation.operation.name}")
    operation.result()  # 等待更新完成
    print("环境更新成功")

# 替换为你的项目ID、区域和环境名称
update_composer_v2_environment("your-project-id", "us-central1", "my-composer-v2-env")

关键说明 #

• 所有示例基于官方google-cloud-composer客户端库，兼容Composer V2；核心差异在于创建环境时需指定environment_type="COMPOSER_V2"。
• Airflow版本、机器类型等参数需符合Google Cloud Composer V2的支持列表，可根据实际需求调整。
• 操作耗时较长（创建/更新通常需10-30分钟），生产环境建议异步处理，避免阻塞主线程。

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