Cloud Run JobsでXに投稿！簡単バッチ処理の実践ガイド

こやまん（エンジニア）

Cloud Run JobsでXに投稿！簡単バッチ処理の実践ガイド

こやまん（エンジニア）

2025.01.16

IT技術

CloudRunJob CloudRunJobs GoogleCloud X

Cloud Run...Jobsに触れてみた！！

皆さんこんにちは。
Google Cloudが好きな小山こと"こやまん"です。
今回は、バッチ処理や非同期処理で活躍するCloud Run Jobsについて触れていきます。

Cloud Run はユーザーがアクセスするアプリケーションに適していますが、Cloud Run Jobs はそれとは異なり、バッチ処理などバックグラウンドで動作させるタスクに適しています。

今回の構築環境

本記事では、下表のような環境を前提に、執筆しています。

OS	macOS Sonoma 14.6.1
Google Cloud	登録済み
Google Cloud CLI	導入済み
X Developer Portal	無料アカウント登録済み（読み書きアプリケーション作成済み）

では早速、はじめていきましょう！

アプリケーションの構築

今回は、X（旧：Twitter）でポストを行うPythonのコードを実装して、これをCloud Run Jobs上で動かしてみます。

main.py

1import os
2import tweepy
3
4def get_twitter_client():
5    """
6    環境変数から認証情報を取得し、Tweepyクライアントを作成して返す
7    """
8    return tweepy.Client(
9        bearer_token=os.getenv('X_BEARER_TOKEN'),
10        consumer_key=os.getenv('X_API_KEY'),
11        consumer_secret=os.getenv('X_API_KEY_SECRET'),
12        access_token=os.getenv('X_ACCESS_TOKEN'),
13        access_token_secret=os.getenv('X_ACCESS_TOKEN_SECRET'),
14    )
15
16
17def post_tweet(client, content):
18    """
19    指定された内容をポスト
20    """
21    try:
22        response = client.create_tweet(text=content)
23        print(f"ポスト成功: {response}")
24    except tweepy.TweepyException as e:
25        print(f"ポスト失敗: {e}")
26
27
28def main():
29    """
30    環境変数からポスト内容を取得し、ポストを投稿
31    """
32    # ポスト内容を取得（環境変数が未設定の場合はデフォルト値を使用）
33    tweet_content = os.getenv("X_POST_TEXT", "デフォルトのポストです。")
34
35    # Twitterクライアントを作成
36    client = get_twitter_client()
37
38    # ポストを投稿
39    post_tweet(client, tweet_content)
40
41
42if __name__ == "__main__":
43    main()

requirements.txt

1# Twitter（X) API を使用するためのライブラリ
2tweepy==4.14.0

Dockerfile

1# ベースイメージ（軽量なPythonイメージを使用）
2FROM python:3.11-slim
3
4# 作業ディレクトリを設定
5WORKDIR /app
6
7# ローカルファイルをコンテナにコピー
8COPY . .
9
10# 必要なPythonパッケージをインストール
11RUN pip install --no-cache-dir -r requirements.txt
12
13# デフォルトで実行するコマンドを設定
14CMD ["python", "main.py"]

Developer Portalより無料プランで登録すれば、各種トークンやキーは取得可能です。
トークン等は、流出した場合悪用されてしまうセキュアな値のため、コードに直書きせず、環境変数より取得するようにしています。

手動でデプロイしてみる

早速、アプリケーションをCloud Run Jobsにデプロイして行ってみましょう。
最初に、Google Cloud プロジェクトにアクセスするために使用するアカウントで認証を行います。
ターミナルで下記コマンドを実行します。

1gcloud auth login

続いて、対象のプロジェクトを設定します。[プロジェクトID]をご自身のプロジェクトIDを設定してください。

1gcloud config set project [プロジェクトID]

設定が完了したら、次のコマンドで操作対象となるプロジェクトの確認を行います。

1gcloud config list project

Cloud Run Jobsにデプロイする際に、Artifact Registryと呼ばれるコンテナイメージ管理サービスを利用するため、事前にリポジトリを作成します。

1gcloud artifacts repositories create x-post \
2    --repository-format=docker \
3    --location=asia-northeast1 \
4    --description="Xへポストするアプリケーションを管理するためのリポジトリ"

Dockerイメージをビルドして、Artifact Registryにプッシュ可能な形式でタグ付けを行います。

1docker build -t asia-northeast1-docker.pkg.dev/[プロジェクトID]/x-post/x-post:latest .
2注：M1、M2の方は、こちらでビルド
3docker buildx build --platform linux/amd64 -t asia-northeast1-docker.pkg.dev/[プロジェクトID]/x-post/x-post:latest .

続いて、Artifact Registryにプッシュ

1docker push asia-northeast1-docker.pkg.dev/[プロジェクトID]/x-post/x-post:latest

そしてCloud Run Jobsにデプロイします。

1gcloud run jobs create x-post \
2--image=asia-northeast1-docker.pkg.dev/[プロジェクトID]/x-post/x-post:latest \
3--region=asia-northeast1

今回は、手動で実行しましたが、継続的にデプロイする場合は、CloudBuildを利用した自動デプロイを推奨します。

Secret Managerの設定

XのDeveloper Portal上で取得したセキュアな値をSecret Managerに登録します。

X_BEARER_TOKEN	Bearer Token の値
X_API_KEY	API Key の値
X_API_KEY_SECRET	API Key Secret の値
X_ACCESS_TOKEN	Access Token の値
X_ACCESS_TOKEN_SECRET	Access Token Secret の値

X_POST_TEXTは、セキュアな値ではないため、Cloud Run Jobs側で直接設定します。

CloudRunJobsの設定

コンソールのジョブの構成の表示と編集より、環境変数とシークレット環境変数を設定します。
CloudRunJobsの環境変数設定

CloudRunJobsの実行

Cloud Run Jobsのジョブは、さまざまな実行方法がありますが、今回はシンプルに手動で実行してみます。
Cloud Run Jobsの管理画面上に実行ボタンがあるので、これをクリックするだけです。

Xへのポスト
無事、ポストされていることが確認できました。手動実行するだけであれば、普通にポストした方が早いですね（笑）

CloudRunJobsの環境変数

Cloud Run Jobsは最大1万件まで並列実行することが可能です。ここで重要となってくる環境変数が2つ存在します。
こちらの2つの環境変数は、Cloud Run Jobsからデフォルトで渡される環境変数で、それぞれのタスクの情報を識別するために使用できます。

CLOUD_RUN_TASK_INDEX	インスタンスのINDEX
CLOUD_RUN_TASK_COUNT	並列実行数

先ほど実装したmain.pyのmainメソッドのみ以下のように修正します。
main.py

1def main():
2    """
3    環境変数からポスト内容を取得し、ポストを投稿
4    """
5    # タスク情報を取得
6    task_index = os.getenv("CLOUD_RUN_TASK_INDEX", "0")
7    task_count = os.getenv("CLOUD_RUN_TASK_COUNT", "1")
8
9    # ポスト内容を作成
10    base_tweet_content = os.getenv("X_POST_TEXT", "デフォルトのポストです。")
11    tweet_content = f"{base_tweet_content} (Task {task_index}/{task_count})"
12
13    # Twitterクライアントを作成
14    client = get_twitter_client()
15
16    # ポストを投稿
17    post_tweet(client, tweet_content)

Dockerイメージをビルドして、Artifact Registryにプッシュ可能な形式でタグ付けを行います。

1docker build -t asia-northeast1-docker.pkg.dev/[プロジェクトID]/x-post/x-post:latest .
2注：M1、M2の方は、こちらでビルド
3docker buildx build --platform linux/amd64 -t asia-northeast1-docker.pkg.dev/[プロジェクトID]/x-post/x-post:latest .

続いて、Artifact Registryにプッシュ

1docker push asia-northeast1-docker.pkg.dev/[プロジェクトID]/x-post/x-post:latest

並列数を３としてCloud Run Jobsにデプロイします。

1gcloud run jobs update x-post \
2    --image=asia-northeast1-docker.pkg.dev/[プロジェクトID]/x-post/x-post:latest \
3    --region=asia-northeast1 \
4    --tasks=3

デプロイ後に再度JOBを実行すると3回ポストされ、それぞれのJOBにインデックスが渡っていることが確認できます。
注：2024年11月現在、Xの無料枠では1日あたり17件のAPI制限がかかっているため、ご注意ください。

CloudRunJobsの活用

今回の記事では、CloudRunJobsを手動で実行しましたが、Pub/SubやCloudSchedulerを利用することで、イベントでのトリガーや定期的な実行によるポストが可能になります。
そして、当然ではありますが、Xへのポスト以外もあらゆるバックグラウンド処理に利用できるため、活用方法は無限大です。
ぜひ、皆さんもCloud Run Jobsを活用して、素晴らしいシステムを構築してみてください。

ITフリーランスのエンジニア専門エージェント