在 Cloud Run 中构建并创建 Python 作业

了解如何创建简单的 Cloud Run 作业，从源代码进行部署，以自动将代码打包到容器映像中，将容器映像上传到 Artifact Registry，然后部署到 Cloud Run。除了所示语言之外，您还可以使用其他语言。

须知事项

  1. 登录您的 Google Cloud 账号。如果您是 Google Cloud 新手，请创建一个账号来评估我们的产品在实际场景中的表现。新客户还可获享 $300 赠金，用于运行、测试和部署工作负载。

  2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  3. 确保您的 Google Cloud 项目已启用结算功能。

  4. 安装 Google Cloud CLI。

  5. 如需初始化 gcloud CLI，请运行以下命令： 

    gcloud init
    6.

  6. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

    Go to project selector

  7. 确保您的 Google Cloud 项目已启用结算功能。

  8. 安装 Google Cloud CLI。

  9. 如需初始化 gcloud CLI，请运行以下命令： 

    gcloud init
    10.

编写示例作业

如需使用 Python 编写作业，请执行以下操作：

  1. 创建名为 jobs 的新目录，并转到此目录中：

    mkdir jobs
cd jobs

  2. 为实际作业代码创建一个 main.py 文件。将以下示例行复制到其中：

    import json
import os
import random
import sys
import time

# Retrieve Job-defined env vars
TASK_INDEX = os.getenv("CLOUD_RUN_TASK_INDEX", 0)
TASK_ATTEMPT = os.getenv("CLOUD_RUN_TASK_ATTEMPT", 0)
# Retrieve User-defined env vars
SLEEP_MS = os.getenv("SLEEP_MS", 0)
FAIL_RATE = os.getenv("FAIL_RATE", 0)

# Define main script
def main(sleep_ms=0, fail_rate=0):
    """Program that simulates work using the sleep method and random failures.

    Args:
        sleep_ms: number of milliseconds to sleep
        fail_rate: rate of simulated errors
    """
    print(f"Starting Task #{TASK_INDEX}, Attempt #{TASK_ATTEMPT}...")
    # Simulate work by waiting for a specific amount of time
    time.sleep(float(sleep_ms) / 1000)  # Convert to seconds

    # Simulate errors
    random_failure(float(fail_rate))

    print(f"Completed Task #{TASK_INDEX}.")

def random_failure(rate):
    """Throws an error based on fail rate

    Args:
        rate: a float between 0 and 1
    """
    if rate < 0 or rate > 1:
        # Return without retrying the Job Task
        print(
            f"Invalid FAIL_RATE env var value: {rate}. "
            + "Must be a float between 0 and 1 inclusive."
        )
        return

    random_failure = random.random()
    if random_failure < rate:
        raise Exception("Task failed.")

# Start script
if __name__ == "__main__":
    try:
        main(SLEEP_MS, FAIL_RATE)
    except Exception as err:
        message = (
            f"Task #{TASK_INDEX}, " + f"Attempt #{TASK_ATTEMPT} failed: {str(err)}"
        )

        print(json.dumps({"message": message, "severity": "ERROR"}))
        sys.exit(1)  # Retry Job Task by exiting the process

    Cloud Run 作业允许用户指定作业要执行的任务数量。此示例代码演示了如何使用内置 CLOUD_RUN_TASK_INDEX 环境变量。每个任务代表容器的一个正在运行的副本。 请注意，任务通常并行执行。如果每个任务都可以独立处理一部分数据，则使用多个任务非常有用。

    每个任务都知道其存储在 CLOUD_RUN_TASK_INDEX 环境变量中的索引。内置 CLOUD_RUN_TASK_COUNT 环境变量包含在执行作业时通过 --tasks 参数提供的任务数量。

    此处所示的代码还展示了如何使用内置 CLOUD_RUN_TASK_ATTEMPT 环境变量重试任务，该变量包含此任务重试的次数，从 0 开始（表示第一次尝试），然后每次连续重试时递增 1，上限为 --max-retries。

    此外，通过代码，您还可以生成失败来测试重试并生成错误日志，从而查看失败情况。

  3. 创建一个名为 Procfile 且无文件扩展名的文本文件，其中包含以下内容：

    web: python3 main.py

您的代码已完成，可以封装在容器中。

构建作业容器，将其发送到 Artifact Registry 并部署到 Cloud Run

重要提示：本快速入门假定您在快速入门中使用的项目中拥有所有者或编辑者角色。否则，请参阅 Cloud Run 部署权限、Cloud Build 权限和 Artifact Registry 权限，获取需要的权限。

本快速入门使用“从源代码部署”，以构建容器，将其上传到 Artifact Registry，然后将作业部署到 Cloud Run：

gcloud run jobs deploy job-quickstart \
    --source . \
    --tasks 50 \
    --set-env-vars SLEEP_MS=10000 \
    --set-env-vars FAIL_RATE=0.1 \
    --max-retries 5 \
    --region REGION \
    --project=PROJECT_ID

其中，PROJECT_ID 是您的项目 ID，REGION 是您的区域，例如 us-central1。请注意，您可以将各种参数更改为要用于测试的任何值。SLEEP_MS 模拟工作，FAIL_RATE 导致 X% 的任务失败，因此您可以试验并行情况并重试失败任务。

在 Cloud Run 中执行作业

如需执行刚刚创建的作业，请运行以下命令：

gcloud run jobs execute job-quickstart --region REGION

将 REGION 替换为您在创建和部署作业时使用的区域，例如 us-central1。

后续步骤

如需详细了解如何使用代码源构建容器并推送到代码库，请参阅：