Google Cloud Run 运算符¶

Cloud Run 用于在完全托管的平台上构建和部署用任何语言（包括 Go、Python、Java、Node.js、.NET 和 Ruby）编写的可扩展容器化应用。

有关该服务的更多信息，请访问 Google Cloud Run 文档。

先决任务¶

要使用这些运算符，您必须执行以下操作

使用 Cloud 控制台选择或创建 Cloud Platform 项目。

为您的项目启用帐单，如 Google Cloud 文档中所述。

启用 API，如 Cloud 控制台文档中所述。
通过 pip 安装 API 库。
pip install 'apache-airflow[google]'
有关安装的详细信息，请访问。
设置 Google Cloud 连接.

创建作业¶

在 Cloud Run 中创建作业之前，您需要定义它。有关作业对象字段的更多信息，请访问 Google Cloud Run 作业说明

简单的作业配置如下所示

tests/system/providers/google/cloud/cloud_run/example_cloud_run.py[源代码]

def _create_job():
    job = Job()
    container = k8s_min.Container()
    container.image = "us-docker.pkg.dev/cloudrun/container/job:latest"
    job.template.template.containers.append(container)
    return job

使用此配置，我们可以创建作业： CloudRunCreateJobOperator

tests/system/providers/google/cloud/cloud_run/example_cloud_run.py[源代码]

create1 = CloudRunCreateJobOperator(
    task_id=create1_task_name,
    project_id=PROJECT_ID,
    region=region,
    job_name=job1_name,
    job=_create_job(),
    dag=dag,
)

请注意，此运算符仅创建作业而不执行它。作业的词典表示将推送到 XCom。

执行作业¶

要执行作业，可以使用

CloudRunExecuteJobOperator

tests/system/providers/google/cloud/cloud_run/example_cloud_run.py[源代码]

execute1 = CloudRunExecuteJobOperator(
    task_id=execute1_task_name,
    project_id=PROJECT_ID,
    region=region,
    job_name=job1_name,
    dag=dag,
    deferrable=False,
)

或者可以在可延迟模式中定义相同的运算符

CloudRunExecuteJobOperator

tests/system/providers/google/cloud/cloud_run/example_cloud_run.py[源代码]

execute2 = CloudRunExecuteJobOperator(
    task_id=execute2_task_name,
    project_id=PROJECT_ID,
    region=region,
    job_name=job2_name,
    dag=dag,
    deferrable=True,
)

还可以指定覆盖，允许你向作业提供新的入口点命令等

CloudRunExecuteJobOperator

tests/system/providers/google/cloud/cloud_run/example_cloud_run.py[源代码]

overrides = {
    "container_overrides": [
        {
            "name": "job",
            "args": ["python", "main.py"],
            "env": [{"name": "ENV_VAR", "value": "value"}],
        }
    ],
    "task_count": 1,
    "timeout": "60s",
}

execute3 = CloudRunExecuteJobOperator(
    task_id=execute3_task_name,
    project_id=PROJECT_ID,
    region=region,
    overrides=overrides,
    job_name=job3_name,
    dag=dag,
    deferrable=False,
)

更新作业¶

要更新作业，可以使用

CloudRunUpdateJobOperator

tests/system/providers/google/cloud/cloud_run/example_cloud_run.py[源代码]

update_job1 = CloudRunUpdateJobOperator(
    task_id=update_job1_task_name,
    project_id=PROJECT_ID,
    region=region,
    job_name=job1_name,
    job=_create_job_with_label(),
    dag=dag,
)

作业的词典表示将推送到 XCom。

列出作业¶

要列出作业，可以使用

CloudRunListJobsOperator

tests/system/providers/google/cloud/cloud_run/example_cloud_run.py[源代码]

list_jobs = CloudRunListJobsOperator(
    task_id=list_jobs_task_name, project_id=PROJECT_ID, region=region, dag=dag
)

运算符采用两个可选参数：“limit”以限制返回的任务数，以及“show_deleted”以在结果中包含已删除的作业。

删除作业¶

要删除作业，可以使用

CloudRunDeleteJobOperator

tests/system/providers/google/cloud/cloud_run/example_cloud_run.py[源代码]

delete_job1 = CloudRunDeleteJobOperator(
    task_id="delete-job1",
    project_id=PROJECT_ID,
    region=region,
    job_name=job1_name,
    dag=dag,
    trigger_rule=TriggerRule.ALL_DONE,
)

请注意，此运算符会等待作业被删除，并且已删除作业的词典表示将推送到 XCom。