完美使用 Google Cloud Run：为单个作业中的每个流程创建新的执行 ID

我目前正在使用 Prefect 云来编排不同的任务，并且我正在使用 CloudRun Push 工作池将这些任务作为作业部署在 Google Cloud Run 上。

这是我面临的情况：每次运行流程时，它都会创建一个新的任务在 Cloud Run 上工作。然而，这不是我想要的行为。相反，我希望每个部署都有一个作业，具有部署名称，并且每个流程运行都应在该作业中创建一个新的执行 ID。

我已经查看了 Prefect 和 Google Cloud Run 文档，但还没有没有找到实现此目的的明确方法。

有人有这种设置的经验或者可以为我指出正确的方向吗？任何帮助将不胜感激！

提前感谢您。

我尝试在 prefect.yaml 中使用以下配置：

  infra_overrides:
    env:
      PREFECT_FLOW_NAME: prefect-daily-testrun
  infrastructure:
    job_variable_name: PREFECT_FLOW_NAME

但结果是相同的：创建了一个新作业，其名称为流动。

你遇到的问题是 Prefect Cloud Run 任务运行器每次运行流程都会创建一个新的 Cloud Run 作业。你希望每个部署有一个作业，并且每次流程运行都会在该作业中创建一个新的执行 ID。

你目前的配置尝试使用 PREFECT_FLOW_NAME 作为作业变量名，但它不起作用，因为 Cloud Run 任务运行器不会将此变量传播到作业创建逻辑。

为了实现你想要的行为，你需要直接使用 Cloud Run API 或 gcloud 命令行工具创建 Cloud Run 作业。然后，你可以使用 CloudRunJob 运行器来触发现有作业上的执行。

以下是实现此目标的步骤：

1. 创建 Cloud Run 作业：

你可以使用 gcloud 命令行工具创建 Cloud Run 作业：

gcloud run jobs create your-job-name \
  --image=us-docker.pkg.dev/cloudrun/container/hello \
  --region=your-region

将 your-job-name 、 image 和 region 替换为你的实际值。

2. 使用 CloudRunJob 运行器：

在你的 Prefect 流程中，使用 CloudRunJob 运行器并指定你创建的作业名称：

from prefect import flow
from prefect_gcp.cloud_run import CloudRunJob

@flow
def my_flow():
    # Your flow logic here

    cloud_run_job_runner = CloudRunJob(
        job_name="your-job-name",
        location="your-region",
    )

    cloud_run_job_runner.submit(wait=False)

if __name__ == "__main__":
    my_flow()

这将使用你之前创建的作业名称触发 Cloud Run 作业上的新执行。每次流程运行都会在现有作业中创建一个新的执行 ID。

注意：

• 确保你拥有创建和管理 Cloud Run 资源的必要权限。
• 你可以使用其他 CloudRunJob 运行器参数，例如 service_account 、 env 等来配置你的执行。

通过这些步骤，你就可以为每个部署创建一个 Cloud Run 作业，并使用 Prefect 在该作业中触发新的执行。