Cloud Run FunctionとCloud Schedulerを使ってETLをする

前書き

本日は【札幌現地+オンライン開催】力強くブログを108記事アウトプットというイベントで拘束されに来ています。とりあえず１本目は書いたのでこの2本目で脱走しようかなと思っています笑

Firebase AnalyticsとBigQuery

モバイルアプリでFirebase Analytics（以下、FA）を使いアプリ計測を行なっている例はよくあります。FAで定義したイベントやユーザープロパティーの設定はFAのダッシュボードやGoogle Analyticsで分析することができます。
より詳細にそれぞれのイベントのCVRを分析する場合にはクエリを書くことになるでしょう。FAではBigQueryとの連携ができ、FAにたまったイベントなどがBigQueryにストリーム転送されます。BigQueryにはFAだけでなくFCMなどの履歴などを連携することができます。

外部データソースをどうBigQueryに取り込むか

モバイルアプリ開発でFirebaseを使っているとBigQueryにデータを集約するのが分析を行う上で都合が良いです。が、バックエンドに別のデータがある場合はそのデータをBigQueryに取り込む必要があります。BigQueryには標準でいくつかのデータソース、例えばAmazon S3やMySqlなどのデータ転送をサポートしています。
ですがすべてのデータソースに対応しているわけではないですし、データソースの構成によっては転送できないこともあります。例えばMySqlですと踏み台となるサーバーを経由したSSLで転送する方法をサポートしているのですが、セキュリティーのために踏み台を用意していないケースなんかがあるでしょう。
ではETL（Extract/Transform/Load）のデータウェアハウスとしてどうBigQueryを運用したら良いでしょうか？

Cloud Run Function

そんな時に使えるのがCloud Run Functionです。一言で言うと「サーバーレスの関数実行環境」です。AWSのLambdaみたいなものといえば良いでしょうか。
同じGoogle Cloudプロジェクト内では認証設定を自動で行ってくれたり、BigQueryやCloud Storageなどの他のリソースへのアクセスが用意なのでデータの抽出や転送がとても実装しやすいです。

トリガーをPub/Subにする

Functionの実行をディスパッチするトリガーがいくつかありますが、おすすめはPub/Subです。のちに説明するCloud Schedulerによる定期実行やBigQuery実行結果のコールバックなどを運用する場合に相性が良いです。Functionで外部ソースからのデータ転送を記述します。Node.jsやPythonなど一般的なサーバー言語で記述できますし、BigQueryのDMLクエリをFunction上で実行するためのライブラリが豊富です。先ほどの例ですと、「Amazon RDSとの間をOIDC認証でより安全にデータソースにアクセスして必要なデータをDMLクエリで直接BigQueryのテーブルに追加する」みたいなユースケースが考えられます。
例えば以下のようなコードを実行することができます。

index.py

from google.cloud import bigquery
import base64
import json

def execute_dml(event, context):
    # Pub/Subメッセージのデコード
    pubsub_message = base64.b64decode(event['data']).decode('utf-8')
    data = json.loads(pubsub_message)

    client = bigquery.Client()

    # DMLクエリの作成（例：UPDATE文）
    # ※ SQLインジェクションを防ぐため、クエリパラメータの使用を推奨します
    query = """
        UPDATE `your-project.your_dataset.your_table`
        SET status = @status
        WHERE id = @id
    """
    
    job_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ScalarQueryParameter("status", "STRING", data.get("status")),
            bigquery.ScalarQueryParameter("id", "INTEGER", data.get("id")),
        ]
    )

    try:
        query_job = client.query(query, job_config=job_config)
        query_job.result()  # 完了を待機
        print(f"DML executed successfully: {query_job.num_dml_affected_rows} rows affected.")
    except Exception as e:
        print(f"Error executing DML: {e}")

認証はIAM, シークレットはSecret Manager

OIDCやリソースへのアクセスはIAMで設定します。認証機構を分けることでより安全にシステム構築できるわけです。また外部リソースへのAPI KeyなどはSecret Managerから環境変数として設定することが可能です。
なお、FunctionやSchedulerに適切なIAMロールを設定する必要があります。必要なロール設定は公式ドキュメントに記載されています。リソースごとに確認しましょう。

以下はサンプルコードと設定の例です。

index.js

const functions = require('@google-cloud/functions-framework');

// Register a CloudEvent callback with the Functions Framework that will
// be executed when the Pub/Sub trigger topic receives a message.
functions.cloudEvent('helloPubSub', cloudEvent => {
  // The Pub/Sub message is passed as the CloudEvent's data payload.
  const base64name = cloudEvent.data.message.data;
  const apiKey = process.env.API_KEY;

  const name = base64name
    ? Buffer.from(base64name, 'base64').toString()
    : 'World';

  console.log(`Hello, ${name}!`);
  console.log('Retrieved Secret API Key:', apiKey);
});

Cloud Scheduler

ではこのCloud Run Functionで外部データソースから１日1回データをストリームするということをやってみましょう。

Cloud Run Function側

まずはpub/subでトピックをひとつ作ります。FunctionはこのトピックのSubscriberになります。Functionを作成するときにそのトピックを選択しましょう。

Cloud Scheduler側

次にCloud Scheduler側で実行した時刻にこのトピックがPublishします。時刻の設定はcronと同じなので特に難しいことはないです。

まとめ

パイプラインはこんな感じです。この記事の一部をGeminiに読み込ませるとNano Bananaが作ってくれました。ちょっと疑問符が浮かぶ画像でしたがそのまま載せることにします笑