Cloud Run Functions から BigQuery にデータを書き込む

はじめに

Cloud Run Functuons にデータを渡すことをトリガーとして BigQuery にデータを書き込む機構を作ってみました。
本記事では名前と好きなスポーツに関するデータが Cloud Run Functions に渡されたら、そのデータを BigQuery に格納してみます。

アーキテクチャ

Cloud Run Functions から BigQuery に書き込むシンプルな構成を取っています。
Cloud Run Functions は IAM 認証によって実行できるようにしています。

1. BigQuery の作成

Google Cloud Console や gcloud CLI など、自身が作成しやすい方法で BigQuery データセットを作成します。(本記事では gcloud CLI を利用)
下記コマンドを実行して user というデータセットを作成します。

bq --location=US mk -d users

次に、テーブルを作成します。
まずテーブルのスキーマを JSON ファイルで作成します。

favorite_sports_schema.json

cat > favorite_sports_schema.json << EOF
[
  {
    "name": "name",
    "type": "STRING",
    "mode": "NULLABLE"
  },
  {
    "name": "favorite_sports",
    "type": "STRING",
    "mode": "NULLABLE"
  }
]
EOF

そして、このスキーマを使うように下記コマンドを実行してテーブルを作成します。

bq mk -t users.favorite_sports favorite_sports_schema.json

Google Cloud Console 上で作成する場合は、下記のようにテーブルを作成します。

2. Cloud Run Functions の作成

Google Cloud Console から Cloud Run のページを開き、「関数を作成」から Cloud Run Functions を作成します。それ以外の「作成」ボタンを押した場合でも下図の一番上の項目で一番左 (Functions) を選択すれば、問題ありません。
今回は Service Name を record-favorite-sports に、ランタイムを Python 3.12、認証は 認証が必要 として IAM 認証を選択しています。

この関数が利用する Service Account はデフォルトの Service Account としていますが、商用利用する場合には最小権限のロールのみ付与している Service Account を利用するようにしてください。

上記、を設定したのち作成ボタンを押下するとコードエディタの画面となるので、以下のコードをコピペします。
コード内の YOUR_PROJECT_ID は自信の値にしてください。

requirements.txt

functions-framework==3.*
google-cloud-bigquery

main.py

import functions_framework
from google.cloud import bigquery

@functions_framework.http
def record_favorite_sports(request):
    """Writes user names and favorite sports to BigQuery."""

    request_json = request.get_json(silent=True)
    request_args = request.args
    print("JSON:" + str(request_json))
    print("args:" + str(request_args))

    bq_client = bigquery.Client()
    table_id = "YOUR_PROJECT_ID.users.favorite_flavors" # YOUR_PROJECT_ID は置き換える

    row_to_insert = [
        {"name": request_json["name"],
        "favorite_sports": request_json["favorite_sports"]
        },
    ]

    errors = bq_client.insert_rows_json(table_id, row_to_insert)  # Make an API request.
    if errors == []:
        return {"message": "New row has been added."}
    else:
        return {"message": "Encountered errors while inserting rows: {}".format(errors)}

エントリポイントを record_favorite_sports とした上でデプロイし、デプロイが完了したら準備完了です。

3. 動作確認

Google Cloud Console の右上にある「テスト」ボタンを押下し、データに下記を入力後「Cloud Shellで実行]
を押下して Cloud Shell で実行します。

{
  "name": "Shohei",
  "favorite_sports": "baseball"
}

Cloud Shell で下記のようなアウトプットが出てきたら成功です。

{"message": "New row has been added."}

BigQuery の画面を確認して、データが格納されているか確認します。

さいごに

今回は Cloud Run Functions から BigQuery にデータを書き込む機構を簡易的に作成してみました。
本機構は "データを自動で書き込む" という点においてはいくらでも応用できそうなので、いろんな処理に応用できないか検討してみようと思います。