Cloud Run Jobsの並列処理を試す

# はじめに

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


# Cloud Run Jobsとは

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

https://cloud.google.com/run/docs/resource-model?hl=ja#jobs


## 並列処理

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

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3618319/7604d837-4886-7841-ebbf-c89a497d1a01.png)


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

```python: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をリリースします．

```shell
#!/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]`)．

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3618319/7caee2c7-a1aa-3b7a-719e-64aafbc4f4b1.png)



# 並列処理に変更

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

```python: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)
```


```python: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()
```

ジョブの設定もアップデートします．
```shell
#!/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つ立ち上がりそれぞれで処理をしていることがわかります．

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3618319/4a2027c3-5ec9-4b39-35a5-d3824cb034dc.png)



# まとめ

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


# 参考

- [Cloud Run Jobs](https://cloud.google.com/run/docs/resource-model?hl=ja#jobs)
- [gcloud beta run jobs create](https://cloud.google.com/sdk/gcloud/reference/beta/run/jobs/create)
- [Cloud Run Jobsを解説する](https://medium.com/google-cloud-jp/cloud-run-jobs-c963a7143367)