Cloud Run Jobsの並列処理を試す

はじめに

現在業務で，Cloud Run Jobsを扱っており，並列処理について検討していたので，その内容を記載します．

Cloud Run Jobsとは

Cloud Run JobsはCloud Runの機能の一つで，手動実行や，スケジュール設定による定期実行をすることが可能です．

並列処理

Cloud Run Jobsには，並列処理という機能が備わっています．下図のイメージのように，1つのジョブに対して，複数のタスクを設定でき，それを直列や並列に処理することが可能となっています．

お題

今回はコンテナ内の処理を何とか並列処理にしたいと思います．スクリプト自体は至ってシンプルなものです．

main.py

"""main.py"""

import json

import click

from src.constants import CLOUD_RUN_TASK_INDEX

@click.command()
@click.option(
    "--arg_list",
    type=str,
)
def main(arg_list: str) -> None:
    arg_list = json.loads(arg_list)
    for _target in arg_list:
        for i in range(10):
            print(
                f"TASK_INDEX: {CLOUD_RUN_TASK_INDEX} target: {_target} number {i}"
            )


if __name__ == "__main__":
    main()

Artifact RegistryにDockerイメージを置き，Cloud Run Jobsをリリースします．

#!/usr/bin/env bash
export $(cat .env | xargs)

docker build -t ${COMPONENT}:latest -f Dockerfile .
IMAGE_NAME=${REGION}-docker.pkg.dev/${PROJECT_ID}/${REGISTRY_NAME}/${COMPONENT}

docker tag ${COMPONENT}:latest ${IMAGE_NAME}:${TAG}
docker push ${IMAGE_NAME}:${TAG}

gcloud beta run jobs create ${JOB_NAME} \
    --project=${PROJECT_ID} \
    --region=${REGION} \
    --image=${IMAGE_NAME} \
    --args=${ARGS} \
    --max-retries=0 \

上記のジョブを実行すると，以下のような結果が得られます(arg_list = [111, 222])．

並列処理に変更

上記コードを，arg_listごと並列処理するものに変更しようと思います．上記コードにも書いていますが，Cloud Run Jobsで処理をすると，環境変数CLOUD_RUN_TASK_INDEXが割り当てられます．これが並列処理ごとのindexとなります．なので，これとarg_listの紐づけを行なって，内部処理を並列にしました．

constants.py

"""constants.py"""
import os

CLOUD_RUN_TASK_INDEX = os.getenv("CLOUD_RUN_TASK_INDEX")
if CLOUD_RUN_TASK_INDEX is not None:
    CLOUD_RUN_TASK_INDEX = int(CLOUD_RUN_TASK_INDEX)

main.py

"""main.py"""

import json

import click

from src.constants import CLOUD_RUN_TASK_INDEX

@click.command()
@click.option(
    "--arg_list",
    type=str,
)
def main(arg_list: str) -> None:
    arg_list = json.loads(arg_list)
    _target = arg_list[CLOUD_RUN_TASK_INDEX]
    for i in range(10):
        print(
            f"TASK_INDEX: {CLOUD_RUN_TASK_INDEX} target: {_target} number {i}"
        )

if __name__ == "__main__":
    main()

ジョブの設定もアップデートします．

#!/usr/bin/env bash
export $(cat .env | xargs)

gcloud beta run jobs update ${JOB_NAME} \
    --project=${PROJECT_ID} \
    --region=${REGION} \
    --image=${IMAGE_NAME} \
    --args=${ARGS} \
    --max-retries=0 \
    --tasks=2 \      # <---- 追加項目
    --parallelism=2  # <---- 追加項目

これで実行すると，以下のように，コンテナが2つ立ち上がりそれぞれで処理をしていることがわかります．

まとめ

今回はCloud Run Jobsの並列処理を，内部のスクリプトに反映しました．内部の処理が簡素すぎるため，並列処理の旨味を感じることができていません．また，ベストプラクティスではない気はしていますが，なにか参考になれば幸いです．

参考

• Cloud Run Jobs
• gcloud beta run jobs create
• Cloud Run Jobsを解説する