Google Cloud Run Operator¶
Cloud Run 用於在完全託管的平臺上構建和部署用任何語言（包括 Go、Python、Java、Node.js、.NET 和 Ruby）編寫的可擴充套件容器化應用程式。
有關此服務的更多資訊，請訪問 Google Cloud Run 文件。
前提任務¶
要使用這些 operator，您必須執行以下操作：
使用 Cloud Console 選擇或建立 Cloud Platform 專案。
啟用專案的結算功能，如 Google Cloud 文件中所述。
啟用 API，如 Cloud Console 文件中所述。
透過 pip 安裝 API 庫。
pip install 'apache-airflow[google]'詳細資訊請參見 安裝。
建立作業¶
在 Cloud Run 中建立作業之前，您需要先定義它。有關 Job 物件欄位的更多資訊，請訪問 Google Cloud Run Job 描述
一個簡單的作業配置可以使用 Job 物件建立
tests/system/google/cloud/cloud_run/example_cloud_run.py
def _create_job_instance() -> Job:
"""
Create a Cloud Run job configuration with google.cloud.run_v2.Job object.
As a minimum the configuration must contain a container image name in its template.
The rest of the configuration parameters are optional and will be populated with default values if not set.
"""
job = Job()
container = k8s_min.Container()
container.image = "us-docker.pkg.dev/cloudrun/container/job:latest"
container.resources.limits = {"cpu": "2", "memory": "1Gi"}
job.template.template.containers.append(container)
return job
或者使用 Python 字典
tests/system/google/cloud/cloud_run/example_cloud_run.py
def _create_job_dict() -> dict:
"""
Create a Cloud Run job configuration with a Python dict.
As a minimum the configuration must contain a container image name in its template.
"""
return {
"template": {
"template": {
"containers": [
{
"image": "us-docker.pkg.dev/cloudrun/container/job:latest",
"resources": {
"limits": {"cpu": "1", "memory": "512Mi"},
"cpu_idle": False,
"startup_cpu_boost": False,
},
"name": "",
"command": [],
"args": [],
"env": [],
"ports": [],
"volume_mounts": [],
"working_dir": "",
"depends_on": [],
}
],
"volumes": [],
"execution_environment": 0,
"encryption_key": "",
},
"labels": {},
"annotations": {},
"parallelism": 0,
"task_count": 0,
},
"name": "",
"uid": "",
"generation": "0",
"labels": {},
"annotations": {},
"creator": "",
"last_modifier": "",
"client": "",
"client_version": "",
"launch_stage": 0,
"observed_generation": "0",
"conditions": [],
"execution_count": 0,
"reconciling": False,
"satisfies_pzs": False,
"etag": "",
}
您可以使用以下任何配置建立 Cloud Run 作業： CloudRunCreateJobOperator
tests/system/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_instance(),
dag=dag,
)
請注意，此 operator 僅建立作業而不執行它。作業的字典表示形式會推送到 XCom。
建立服務¶
在 Cloud Run 中建立服務之前，您需要先定義它。有關 Service 物件欄位的更多資訊，請訪問 Google Cloud Run Service 描述
一個簡單的服務配置如下所示
tests/system/google/cloud/cloud_run/example_cloud_run_service.py
def _create_service():
service = Service()
container = k8s_min.Container()
container.image = "us-docker.pkg.dev/cloudrun/container/placeholder:latest"
service.template.containers.append(container)
return service
使用此配置，我們可以建立服務： CloudRunCreateServiceOperator
tests/system/google/cloud/cloud_run/example_cloud_run_service.py
create_cloud_run_service = CloudRunCreateServiceOperator(
task_id="create-cloud-run-service",
project_id=PROJECT_ID,
region="us-central1",
service=_create_service(),
service_name="cloudrun-system-test-service",
)
請注意，此 operator 僅建立服務而不執行它。服務的字典表示形式會推送到 XCom。
刪除服務¶
使用此配置，我們可以刪除服務： CloudRunDeleteServiceOperator
tests/system/google/cloud/cloud_run/example_cloud_run_service.py
delete_cloud_run_service = CloudRunDeleteServiceOperator(
task_id="delete-cloud-run-service",
project_id=PROJECT_ID,
region="us-central1",
service_name="cloudrun-system-test-service",
dag=dag,
)
請注意，此 operator 會等待服務刪除完成，已刪除服務的字典表示形式會推送到 XCom。
執行作業¶
要執行作業，您可以使用
tests/system/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,
)
或者您可以以 deferrable 模式定義相同的 operator
tests/system/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,
)
您還可以指定覆蓋（overrides），允許您為作業指定新的入口點命令等。
tests/system/google/cloud/cloud_run/example_cloud_run.py
overrides = {
"container_overrides": [
{
"name": "job",
"args": ["python", "main.py"],
"env": [{"name": "ENV_VAR", "value": "value"}],
"clear_args": False,
}
],
"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,
)
更新作業¶
要更新作業，您可以使用
tests/system/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_instance_with_label(),
dag=dag,
)
作業的字典表示形式會推送到 XCom。
列出作業¶
要列出作業，您可以使用
tests/system/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
)
此 operator 接受兩個可選引數：“limit” 用於限制返回的任務數量，“show_deleted” 用於在結果中包含已刪除的作業。
刪除作業¶
要刪除作業，您可以使用
tests/system/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,
)
請注意，此 operator 會等待作業刪除完成，已刪除作業的字典表示形式會推送到 XCom。