はじめに：Meta広告だけDTSが無く手作業になりがちな課題

Google広告には BigQuery Data Transfer Service（DTS）があり、設定さえ済ませれば毎日自動で実績が BigQuery に入ってきます。ところが Meta広告（Facebook / Instagram）には、これに相当する公式の「ボタンひとつで BigQuery へ」という仕組みがありません。

そのため、多くの中小ECや個人事業主の方が、

  • 管理画面を開いて期間を指定し
  • CSV をダウンロードして
  • スプレッドシートに貼り付けて
  • Google広告のデータと手で突き合わせる

という作業を、毎週・毎月くり返しているのではないでしょうか。件数が増えるほどミスも増えますし、何より時間がもったいないですよね。

この記事では、Meta広告の Insights API から実績データを取得し、BigQuery に自動でロードする Python スクリプトの作り方を、最小構成で紹介します。一度組んでしまえば、あとはスケジュール実行で「気づいたらデータが溜まっている」状態を目指せます。

この記事はあくまで仕組みづくりの一例です。Meta広告 API の細かい仕様やバージョン番号は頻繁に更新されるため、実装の際は必ず最新の公式ドキュメント（Meta for Developers / Marketing API）を確認してください。

全体像

やることはシンプルで、次の3ステップです。

  1. Meta の Insights API を叩いて、広告の実績（インプレッション・クリック・費用など）を JSON で取得する
  2. 取得したデータを整形する
  3. google-cloud-bigquery で BigQuery のテーブルにロードする

これを Cloud Functions + Cloud Scheduler に乗せれば、毎朝自動で動く「Meta広告 → BigQuery」のパイプラインになります。

前提：Meta側の準備

コードを書く前に、Meta側で以下を用意しておく必要があります。詳細な手順は公式に従ってください。ここでは概要だけ押さえます。

  • Meta for Developers のアプリを作成する
  • アクセストークンを取得する（後述のとおり、短期トークンではなく長期トークン、またはシステムユーザートークンの利用を検討します）
  • アプリに ads_read 権限を付与する（広告実績の取得に必要）
  • 対象の 広告アカウントID（act_ から始まる形式）を控えておく

そして Google Cloud 側では、

  • BigQuery を有効化したプロジェクト
  • データセット（例：meta_ads）
  • スクリプトを動かすサービスアカウント（BigQuery へのロード権限を持つもの）

を準備します。

アクセストークンや認証情報は、コード内に直接書かず、必ず環境変数やシークレットマネージャーで管理してください。GitHub などに誤ってトークンを push すると、広告アカウントが不正利用されるおそれがあります。

Insights API から実績を取得してBigQueryへロードする

それでは実際のコードを見ていきます。ここでは外部 SDK を入れずに済むよう、HTTP リクエストは requests で直接行います（Meta公式の facebook_business SDK を使う方法もあります）。

まず、必要なパッケージをインストールします。

pip install requests google-cloud-bigquery

環境変数には、トークンやアカウントIDなどの秘匿情報を入れておきます。

export META_ACCESS_TOKEN="（あなたの長期アクセストークン）"
export META_AD_ACCOUNT_ID="act_xxxxxxxxxx"
export GCP_PROJECT_ID="your-gcp-project"
export BQ_DATASET="meta_ads"
export BQ_TABLE="ad_insights"

次に、Insights API を呼び出す部分です。

import os
import datetime
import requests

# APIバージョンは最新の公式ドキュメントに合わせて指定する
GRAPH_API_VERSION = "v21.0"

ACCESS_TOKEN = os.environ["META_ACCESS_TOKEN"]
AD_ACCOUNT_ID = os.environ["META_AD_ACCOUNT_ID"]


def fetch_insights(since: str, until: str) -> list[dict]:
    """指定期間のアカウント実績を日別・広告単位で取得する"""
    url = (
        f"https://graph.facebook.com/{GRAPH_API_VERSION}"
        f"/{AD_ACCOUNT_ID}/insights"
    )
    params = {
        "access_token": ACCESS_TOKEN,
        "level": "ad",  # campaign / adset / ad から選ぶ
        "time_increment": 1,  # 1 = 日別
        "time_range": f'{{"since":"{since}","until":"{until}"}}',
        "fields": ",".join([
            "date_start",
            "account_id",
            "campaign_name",
            "adset_name",
            "ad_name",
            "impressions",
            "clicks",
            "spend",
        ]),
        "limit": 500,
    }

    rows: list[dict] = []
    while True:
        res = requests.get(url, params=params, timeout=60)
        res.raise_for_status()
        body = res.json()
        rows.extend(body.get("data", []))

        # ページネーション：次ページがあれば辿る
        next_url = body.get("paging", {}).get("next")
        if not next_url:
            break
        url, params = next_url, {}

    return rows

time_increment=1 を指定すると日別で取れるので、後から BigQuery 側で柔軟に集計できます。level を campaign や adset に変えれば、取得の粒度を調整できます。

取得した JSON は文字列のままなので、BigQuery に入れる前に型をそろえておきます。

def transform(rows: list[dict]) -> list[dict]:
    """API のレスポンスを BigQuery 用に整形する"""
    cleaned = []
    for r in rows:
        cleaned.append({
            "date": r.get("date_start"),
            "account_id": r.get("account_id"),
            "campaign_name": r.get("campaign_name"),
            "adset_name": r.get("adset_name"),
            "ad_name": r.get("ad_name"),
            "impressions": int(r.get("impressions", 0)),
            "clicks": int(r.get("clicks", 0)),
            "spend": float(r.get("spend", 0)),
            "loaded_at": datetime.datetime.utcnow().isoformat(),
        })
    return cleaned

最後に、整形したデータを BigQuery へロードします。ここでは load_table_from_json を使います。

from google.cloud import bigquery

PROJECT_ID = os.environ["GCP_PROJECT_ID"]
DATASET = os.environ["BQ_DATASET"]
TABLE = os.environ["BQ_TABLE"]


def load_to_bigquery(records: list[dict]) -> None:
    if not records:
        print("ロード対象のデータがありません。")
        return

    client = bigquery.Client(project=PROJECT_ID)
    table_id = f"{PROJECT_ID}.{DATASET}.{TABLE}"

    schema = [
        bigquery.SchemaField("date", "DATE"),
        bigquery.SchemaField("account_id", "STRING"),
        bigquery.SchemaField("campaign_name", "STRING"),
        bigquery.SchemaField("adset_name", "STRING"),
        bigquery.SchemaField("ad_name", "STRING"),
        bigquery.SchemaField("impressions", "INTEGER"),
        bigquery.SchemaField("clicks", "INTEGER"),
        bigquery.SchemaField("spend", "FLOAT"),
        bigquery.SchemaField("loaded_at", "TIMESTAMP"),
    ]
    job_config = bigquery.LoadJobConfig(
        schema=schema,
        write_disposition="WRITE_APPEND",  # 追記。洗い替えなら WRITE_TRUNCATE
    )

    job = client.load_table_from_json(
        records, table_id, job_config=job_config
    )
    job.result()  # 完了まで待つ
    print(f"{len(records)} 行を {table_id} にロードしました。")


def main():
    # 昨日1日分を取得する例
    yesterday = (datetime.date.today() - datetime.timedelta(days=1)).isoformat()
    rows = fetch_insights(since=yesterday, until=yesterday)
    records = transform(rows)
    load_to_bigquery(records)


if __name__ == "__main__":
    main()

これで、ローカルから python main.py を実行すれば、昨日分の Meta広告実績が BigQuery に入ります。まずは手元で動かして、テーブルに行が増えることを確認してみてください。

重複を防ぐ工夫

WRITE_APPEND で毎日追記していくと、再実行したときに同じ日付のデータが二重に入ってしまいます。運用では次のどちらかをおすすめします。

  • 取得対象の日付パーティションだけを一度削除してから追記する
  • いったんステージング用テーブルにロードし、MERGE 文で本テーブルへ反映する

日付でパーティション分割したテーブルにしておくと、この管理がぐっと楽になります。

スケジュール化：Cloud Functions + Cloud Scheduler

ローカルで動いたら、次は自動化です。サーバーを常時立てなくても、Google Cloud のマネージドサービスだけで完結できます。

おおまかな流れは次のとおりです。

  1. 上記の main 関数を Cloud Functions（第2世代）としてデプロイする
  2. Cloud Scheduler から、毎朝決まった時刻にその関数を呼び出す

たとえば Cloud Functions のデプロイと、毎朝7時（日本時間）に実行するスケジュールの作成は、概念的には次のようなコマンドになります（実際のフラグやランタイム名は最新の公式に合わせてください）。

# 関数のデプロイ（例）
gcloud functions deploy meta-ads-import \
  --gen2 \
  --runtime=python312 \
  --region=asia-northeast1 \
  --trigger-http \
  --entry-point=main \
  --set-env-vars=GCP_PROJECT_ID=your-gcp-project,BQ_DATASET=meta_ads,BQ_TABLE=ad_insights

# 毎朝7時(JST)に呼び出すスケジュールの作成（例）
gcloud scheduler jobs create http meta-ads-daily \
  --schedule="0 7 * * *" \
  --time-zone="Asia/Tokyo" \
  --uri="（デプロイした関数のURL）" \
  --http-method=POST

アクセストークンのような秘匿情報は、環境変数に直書きするのではなく Secret Manager に格納し、関数から参照する形にするとより安全です。

これで「毎朝、自動的に前日分の Meta広告実績が BigQuery に溜まる」状態ができあがります。Google広告のデータと同じデータセットに集めておけば、媒体をまたいだ横断レポートも作りやすくなります。

注意：トークンの有効期限・レート制限・規約

自動化を組むうえで、いくつか気をつけたい点があります。

アクセストークンの有効期限：管理画面から取得できる短期トークンは数時間で切れてしまいます。自動運用には長期トークンやシステムユーザートークンを使い、定期的な失効・更新の運用も想定しておきましょう。トークンが切れると、ある朝突然データが止まります。

レート制限：Marketing API には呼び出し回数の上限があります。大量のアカウントや細かい粒度で一気に取得するとエラーになることがあります。取得粒度を必要な範囲にとどめ、エラー時はリトライ間隔を空けるなどの配慮をしてください。

規約とデータの取り扱い：取得した広告データの利用は、Meta のプラットフォーム規約・開発者ポリシーの範囲内で行ってください。個人を特定しうる情報の扱いにも注意が必要です。仕様変更も多い領域なので、定期的に公式アナウンスを確認することをおすすめします。

まとめ

Meta広告は公式の自動転送がない分、つい手作業になりがちな媒体です。けれども、

  • Insights API で実績を取得し
  • Python で整形して BigQuery にロードし
  • Cloud Functions + Cloud Scheduler で毎日自動実行する

という流れを一度組んでしまえば、あとは手を離して大丈夫です。Google広告のデータと同じ場所に集めておけば、媒体横断のレポートづくりも一気にラクになります。

まずは「昨日1日分をローカルで BigQuery に入れる」ところから、小さく始めてみてください。

なお、取得したデータを GA4 やクリエイティブ単位の ROAS と組み合わせて分析したい方は Meta広告とGA4をBigQueryで突き合わせてクリエイティブ別ROASを見る を、Google広告側の自動転送をまだ設定していない方は Google広告のBigQuery Data Transferを設定する もあわせてご覧ください。