Cloud Scheduler × Cloud Functions × BigQueryでサーバーレスETLを構築する

はじめに：サーバーを持たずに定期ETLしたい

「広告管理画面のデータや在庫APIの数字を、毎日BigQueryに取り込んで一元管理したい」——中小ECの分析基盤を整えていくと、必ずこの「外部データの定期取り込み」にぶつかります。

一番素直なやり方は、どこかのサーバーで毎日決まった時間にスクリプトを動かすことです。ですが、そのためだけにサーバーを一台借りて、OSをアップデートして、止まっていないか監視して……というのは、規模に対して明らかに重すぎます。動いていない時間も料金は発生しますし、運用の手間も馬鹿になりません。

この記事では、サーバーを一切常駐させずに、外部APIから取得したデータを毎日BigQueryへ流し込む構成を紹介します。使うのは Google Cloud の3つのサービスだけです。実行されるのは1日に数秒から数分で、その間しか課金されません。小規模なEC事業でも無理なく持てる軽さで、定期ETLを組めます。

全体構成：Scheduler → Functions → 外部API → BigQuery

まず、処理の流れを整理します。登場するのは次の3つです。

Cloud Scheduler：決まった時刻に処理を起動するスケジューラ。いわばクラウド版のcronです。
Cloud Functions：起動されたときだけ動く関数。ここで外部APIを叩き、データを整形してBigQueryに書き込みます。
BigQuery：取り込んだデータの保存先。集計や分析はここで行います。

時系列で書くと、こうなります。

Cloud Scheduler が、毎朝決まった時刻に Cloud Functions を呼び出す
Cloud Functions が外部APIにリクエストを送り、データを取得する
取得したデータを整形し、BigQuery のテーブルへロードする
関数が終了する。次の起動時刻まで何も動かず、課金もされない

ポイントは、常に動いているものが何もないことです。Scheduler は時刻が来たら一瞬だけ動き、Functions も呼ばれたときだけ起動します。普段は止まっているので、サーバーの維持費という考え方そのものがなくなります。

この構成は、BigQuery内部のテーブルからテーブルへの加工とは役割が違います。BigQueryに入った後のデータマート更新は、BigQueryのスケジュールクエリでデータマートを毎朝自動更新するで紹介したスケジュールクエリが向いています。今回のテーマは、その手前にある「外部のデータをBigQueryに入れるところ」です。

Cloud Functions（Python）の実装例

では、心臓部である Cloud Functions の中身を見ていきます。ここでは Python で、外部APIからJSONを取得し、それを BigQuery にロードするまでを書きます。

外部APIの認証情報やプロジェクト名は、コードに直接書かず環境変数から読み込みます。トークンをソースに直書きすると、リポジトリに残ってしまったり、共有時に漏れたりと事故のもとになるためです。Cloud Functions では、デプロイ時に環境変数を設定できます。

import os
import json
import datetime
import requests
from google.cloud import bigquery

# 設定値は環境変数から読み込む（コードに直書きしない）
API_ENDPOINT = os.environ["API_ENDPOINT"]
API_TOKEN = os.environ["API_TOKEN"]
PROJECT_ID = os.environ["BQ_PROJECT_ID"]
DATASET = os.environ["BQ_DATASET"]
TABLE = os.environ["BQ_TABLE"]


def fetch_records():
    """外部APIから当日分のデータを取得して整形する"""
    today = datetime.date.today().isoformat()
    headers = {"Authorization": f"Bearer {API_TOKEN}"}
    params = {"date": today}

    response = requests.get(
        API_ENDPOINT, headers=headers, params=params, timeout=30
    )
    response.raise_for_status()
    raw = response.json()

    # BigQueryにロードしやすい形（フラットな辞書のリスト）へ変換
    records = []
    for item in raw.get("data", []):
        records.append(
            {
                "date": today,
                "item_id": str(item["id"]),
                "amount": int(item["amount"]),
                "loaded_at": datetime.datetime.utcnow().isoformat(),
            }
        )
    return records


def load_to_bigquery(records):
    """整形済みレコードをBigQueryのテーブルへロードする"""
    client = bigquery.Client(project=PROJECT_ID)
    table_ref = f"{PROJECT_ID}.{DATASET}.{TABLE}"

    job_config = bigquery.LoadJobConfig(
        write_disposition="WRITE_TRUNCATE",  # 当日分を入れ替える
        schema=[
            bigquery.SchemaField("date", "DATE"),
            bigquery.SchemaField("item_id", "STRING"),
            bigquery.SchemaField("amount", "INTEGER"),
            bigquery.SchemaField("loaded_at", "TIMESTAMP"),
        ],
    )

    job = client.load_table_from_json(
        records, table_ref, job_config=job_config
    )
    job.result()  # ロード完了まで待つ
    return job.output_rows


def main(request):
    """Cloud SchedulerからのHTTPリクエストで起動されるエントリーポイント"""
    records = fetch_records()
    if not records:
        return ("no records", 200)

    rows = load_to_bigquery(records)
    return (f"loaded {rows} rows", 200)

load_table_from_json は、辞書のリストをそのまま BigQuery にロードできるので、外部APIのJSONレスポンスを扱うときに相性が良い書き方です。スキーマを明示しておくと、APIの返す型が揺れても意図しないカラムが作られず、テーブルの構造が安定します。

なお、write_disposition を WRITE_TRUNCATE にすると、ロードのたびにテーブル全体を入れ替えます。日付パーティションのある大きなテーブルに追記していきたい場合は、宛先を TABLE$20260720 のように日付指定にして、そのパーティションだけを入れ替える方法もあります。データ量と用途に応じて選んでください。

このパターンは外部広告データの取り込みにもそのまま応用できます。Meta広告のAPIから取り込む具体例は、Meta広告のデータをAPIとPythonでBigQueryに取り込むで扱っています。

Cloud Scheduler の設定

関数ができたら、それを毎日決まった時刻に呼び出す Cloud Scheduler を設定します。コンソールからも作れますが、再現性を考えるとコマンドで残しておくのがおすすめです。概要は次のとおりです。

# 毎朝6時(JST)にCloud FunctionsのURLをHTTPで叩くジョブを作成
gcloud scheduler jobs create http daily-etl-job \
  --location asia-northeast1 \
  --schedule "0 6 * * *" \
  --time-zone "Asia/Tokyo" \
  --uri "https://REGION-PROJECT.cloudfunctions.net/daily-etl" \
  --http-method GET \
  --oidc-service-account-email "scheduler-invoker@PROJECT.iam.gserviceaccount.com"

--schedule は cron 形式で指定します。0 6 * * * なら毎朝6時です。--time-zone を指定しておかないと UTC 基準で動いてしまい、9時間ずれるので注意してください。

--oidc-service-account-email は、Scheduler が Cloud Functions を呼び出すときの認証に使うサービスアカウントです。関数を一般公開せず、このサービスアカウントからの呼び出しだけを許可しておくと、外部から勝手に叩かれることを防げます。

冪等性・リトライ・エラー通知の注意点

サーバーレスETLは手軽な反面、放置すると「いつの間にか壊れていた」が起きやすい仕組みでもあります。運用で気をつけたい点を3つ挙げます。

冪等性を確保する

同じ処理が二度走っても結果が壊れないようにしておきます。Scheduler のリトライや手動の再実行で、関数が複数回呼ばれることは普通に起こるためです。

先のコードのように当日分を WRITE_TRUNCATE で入れ替える、あるいは日付パーティション単位で置き換える設計にしておけば、何回流しても結果は同じになります。単純な追記（APPEND）だけにすると、再実行のたびに重複行が積み上がってしまうので避けたいところです。

リトライを前提に組む

外部APIは、こちらに非がなくても一時的にエラーを返すことがあります。Cloud Scheduler にはリトライの設定があるので、回数や間隔を指定しておきます。

ただし、リトライが意味を持つのは冪等性が確保できているときだけです。先に冪等性を担保したうえで、リトライを有効にするという順番を守ってください。関数側のタイムアウトも、APIの応答が遅いケースを見越して余裕を持たせておくと安心です。

失敗に気づける仕組みを置く

一番怖いのは、エラーが起きていることに誰も気づかないまま、テーブルの更新が止まっている状態です。最低限、Cloud Functions の実行が失敗したらアラートが飛ぶようにしておきます。

Cloud Monitoring で関数のエラー率に対するアラートを設定し、メールやチャットへ通知する方法が手軽です。あわせて、ロードしたテーブルの loaded_at が当日になっているかを別途チェックすると、「関数は成功扱いだがデータが空だった」という見落としにも気づけます。

まとめ

外部APIのデータをBigQueryへ定期的に取り込む処理は、Cloud Scheduler・Cloud Functions・BigQuery の組み合わせで、サーバーを常駐させずに構築できます。

普段は何も動かないので、維持費と運用の手間が小さい
Functions（Python）で取得・整形し、load_table_from_json でBigQueryへロードする
認証情報は環境変数で渡し、コードに直書きしない
冪等性・リトライ・エラー通知をセットで考えておくと、止まっても気づける

まずは取り込みたい外部データを1つ決めて、小さく動かしてみるのが近道です。動く流れが一度できれば、別のAPIへ横展開するのは難しくありません。