⚡Shopifyの注文Webhook × Cloud Functions × BigQueryでリアルタイム売上基盤を作る

はじめに：日次バッチでは間に合わない場面がある

中小ECの売上分析は、ほとんどの場合「日次バッチで十分」です。前日分の注文を翌朝にまとめてBigQueryへ取り込み、ダッシュボードを更新する——これで困ることはまずありません。実際、定期取り込みの基本形は別記事の Cloud Scheduler × Cloud Functions × BigQueryでサーバーレスETLを構築する で解説したとおりで、まずはこの形を持っておけば困りません。

ですが、年に数回、日次では明らかに遅い場面があります。たとえば次のようなときです。

• セール開始から数時間で、どの商品がどれだけ売れているかを見たい
• テレビやSNSで紹介された直後、トラフィックと注文がどう動いているかを追いたい
• 数量限定の販売で、在庫の減りをほぼリアルタイムに把握したい

こうした場面で「翌朝のレポート」を待つのは現実的ではありません。かといって、管理画面を何度もリロードして手で数えるのも限界があります。欲しいのは、注文が入った瞬間にBigQueryへレコードが積まれ、いつものダッシュボードでそのまま見られる状態です。

この記事では、Shopifyの注文Webhookを Cloud Functions で受け取り、数秒でBigQueryに反映するリアルタイム売上基盤の作り方を解説します。日次バッチを置き換えるものではなく、必要な期間だけ併走させる「速報レーン」として位置づけるのがおすすめです。

なお、ShopifyやGoogle Cloudの管理画面の項目名・API仕様は変更されることがあります。実際に構築する際は、必ず最新の公式ドキュメントで現在の仕様を確認してください。本記事は構成と考え方の説明を主目的としています。

全体構成：Webhook → Cloud Functions → BigQuery

今回作る構成は、登場人物が3つだけのシンプルなものです。

1. Shopifyの注文Webhook：注文が作成されたタイミング（orders/create トピック）で、注文データをHTTPで送信する
2. Cloud Functions：そのHTTPリクエストを受け取り、署名を検証し、必要な項目を整形する
3. BigQuery：整形したレコードを注文テーブルへ挿入する

データの流れを図にすると次のようになります。

[Shopify]
   │  注文が作成される
   │  orders/create Webhook (HTTP POST)
   ▼
[Cloud Functions (HTTPトリガー)]
   │  1. HMAC署名を検証
   │  2. ペイロードから必要項目を抽出・整形
   │  3. BigQueryへ挿入
   ▼
[BigQuery: orders_realtime テーブル]
   │
   ▼
[Looker Studio などのダッシュボード]

ポイントは、Shopify側が「注文が入るたびにこちらへ通知してくれる」という点です。こちらから定期的に取りに行く（ポーリングする）必要がなく、Cloud Functions は通知が来たときだけ起動します。動いている時間しか課金されないため、速報レーンを常設してもコストはごくわずかで済みます。

Webhookの登録について

orders/create のWebhookは、Shopifyの管理画面（設定 → 通知 → Webhook）から登録する方法と、Admin API経由で登録する方法があります。送信先URLには、後述する Cloud Functions のエンドポイントを指定します。

登録時に発行される署名用のシークレット（Webhook signing secret）は、後の署名検証で必須になります。この値は秘密情報なので、コードに直接書かず、環境変数やSecretManagerで管理してください。登録手順や項目名は変わることがあるため、最新の公式ドキュメントを確認しながら進めます。

実装：Cloud Functions で受けてBigQueryへ

ここからは Python での実装例です。HTTPトリガーの Cloud Functions を想定しています。処理の流れは「署名検証 → ペイロード整形 → BigQueryへ挿入」の3段階です。

1. HMAC署名の検証

Webhookは公開URLで受けるため、Shopify以外の誰かが偽のリクエストを送ってくる可能性があります。これを防ぐのが署名検証です。

Shopifyは、リクエストボディを署名用シークレットでHMAC-SHA256計算し、Base64エンコードした値を X-Shopify-Hmac-Sha256 ヘッダーに付けて送ってきます。受け取った側で同じ計算をして、一致するかを確認します。

import base64
import hashlib
import hmac
import os


def verify_webhook(request_body: bytes, hmac_header: str) -> bool:
    """ShopifyのWebhook署名を検証する"""
    secret = os.environ["SHOPIFY_WEBHOOK_SECRET"]  # 環境変数から取得

    digest = hmac.new(
        secret.encode("utf-8"),
        request_body,
        hashlib.sha256,
    ).digest()
    computed_hmac = base64.b64encode(digest).decode("utf-8")

    # タイミング攻撃を避けるため compare_digest を使う
    return hmac.compare_digest(computed_hmac, hmac_header or "")

ここで重要なのは、HMACの計算対象が「生のリクエストボディ（バイト列）」だという点です。JSONをパースしてから文字列化し直すと、空白やキーの順序が変わって署名が一致しなくなることがあります。検証は必ずパース前の生ボディに対して行ってください。

また、hmac.compare_digest を使うことで、単純な文字列比較で起こりうるタイミング攻撃を避けています。

2. ペイロードの整形

署名が確認できたら、JSONをパースして、BigQueryに入れたい項目だけを取り出します。Shopifyの注文ペイロードは非常に項目数が多いので、分析で使うものだけに絞るのが扱いやすいです。

import json
from datetime import datetime, timezone


def build_row(payload: dict) -> dict:
    """注文ペイロードからBigQuery用の1行を組み立てる"""
    return {
        "order_id": str(payload.get("id", "")),
        "order_name": payload.get("name"),          # 例: #1001
        "created_at": payload.get("created_at"),     # Shopify側の注文日時(ISO8601)
        "total_price": payload.get("total_price"),
        "currency": payload.get("currency"),
        "customer_id": str((payload.get("customer") or {}).get("id", "")),
        "line_item_count": len(payload.get("line_items", [])),
        # 取り込み時刻も記録しておくと遅延の確認に役立つ
        "ingested_at": datetime.now(timezone.utc).isoformat(),
    }

金額（total_price）は文字列で届くことがあるため、BigQuery側のカラム型に合わせて扱います。本記事の例では文字列のまま入れ、集計時にキャストする方針にしていますが、数値型カラムへ入れるならここでキャストします。

3. BigQueryへの挿入

整形した行を BigQuery へ挿入します。リアルタイム性を優先するなら、ストリーミング挿入である insert_rows_json がシンプルです。

from google.cloud import bigquery

bq_client = bigquery.Client()

TABLE_ID = os.environ["BQ_TABLE_ID"]  # 例: my-project.ec.orders_realtime


def insert_to_bigquery(row: dict, insert_id: str) -> None:
    # row_ids に webhook_id を渡すと、BigQuery側でベストエフォートの
    # 重複排除が効く(同じ insertId の行は短時間なら弾かれやすくなる)
    errors = bq_client.insert_rows_json(
        TABLE_ID,
        [row],
        row_ids=[insert_id] if insert_id else None,
    )
    if errors:
        # 挿入エラーはログに残し、例外を投げて再送を促す
        raise RuntimeError(f"BigQuery insert errors: {errors}")

insert_rows_json は1件から使え、届いた注文をすぐに反映できるのが利点です。row_ids（内部的には各行の insertId）に webhook_id のような一意キーを渡しておくと、ストリーミング挿入が備えるベストエフォートの重複排除が働き、短時間に再送された同一通知はBigQuery側で弾かれやすくなります。ただしこれはあくまで「ベストエフォート」で、一定時間を超えた再送や挿入経路の事情によっては重複が残ることがあります。完全な排除は保証されないため、後述のとおり集計側でも畳む二段構えにしておくのが安全です。

注文量が非常に多く、ストリーミング挿入の料金や上限が気になる場合は、いったん Cloud Storage や一時バッファへ書き出し、まとめて load（ロードジョブ）で取り込む方式に切り替える選択肢もあります。ロードジョブは無料枠の範囲で扱いやすい一方、反映までに数十秒以上かかることがあり、リアルタイム性とのトレードオフになります。速報レーンとしては、まず insert_rows_json で始めるのが分かりやすいでしょう。料金体系や上限は変わるため、本格運用の前に最新の公式情報を確認してください。

4. 全体を1つの関数にまとめる

ここまでの部品を、HTTPトリガーのエントリポイントから呼び出します。

import functions_framework


@functions_framework.http
def shopify_order_webhook(request):
    # 署名検証(生ボディに対して行う)
    raw_body = request.get_data()
    hmac_header = request.headers.get("X-Shopify-Hmac-Sha256", "")
    if not verify_webhook(raw_body, hmac_header):
        return ("invalid signature", 401)

    # 再送判定に使うWebhook-Idを取得
    webhook_id = request.headers.get("X-Shopify-Webhook-Id", "")

    # ペイロード整形(webhook_idも行に含める)
    payload = json.loads(raw_body)
    row = build_row(payload, webhook_id)

    # BigQueryへ挿入(insertId に webhook_id を渡す)
    insert_to_bigquery(row, webhook_id)

    # 200を返してShopifyに受領を伝える
    return ("ok", 200)

署名が不正なら 401 を返して処理を打ち切ります。正常に取り込めたら 200 を返し、Shopifyに「受け取った」ことを伝えます。この応答コードが、次に説明する再送の挙動と深く関係します。

重複対策：同じ注文が二重に入らないようにする

Webhookの世界では、「同じ通知が複数回届くことがある」という前提で設計する必要があります。ネットワークの一時的な不調などで、Shopify側がレスポンスを受け取れなかったと判断すると、同じWebhookを再送するためです。

何も対策しないと、再送のたびに同じ注文がBigQueryへ重複挿入され、売上が水増しされてしまいます。これを防ぐには、処理を**冪等（べきとう）**にします。つまり「同じ通知を何度受け取っても、結果が1回分にしかならない」状態を作ります。

実用的なのは、Shopifyがリクエストヘッダーに付けてくる X-Shopify-Webhook-Id を一意キーとして使う方法です。

• 取り込んだ webhook_id を行に記録しておく
• 同じ webhook_id がすでに存在する場合は挿入をスキップする

def build_row(payload: dict, webhook_id: str) -> dict:
    row = {
        # ...(前述の項目)...
        "webhook_id": webhook_id,  # 重複判定の一意キー
    }
    return row

対策は1段では決め切らず、挿入時とクエリ時の二段構えにするのが実用的です。

• 挿入時：row_ids（insertId）で弾く：前述のとおり、insert_rows_json の row_ids に webhook_id を渡しておくと、ストリーミング挿入が備えるベストエフォートの重複排除が働き、短時間に再送された同一通知の多くはBigQueryへ届く前に弾かれます。これは標準的な手法で、コードの追加もわずかです。ただし「ベストエフォート」であり、一定時間を超えた再送などでは漏れることがあります
• 集計時：クエリ側で畳む：挿入時の排除が漏れても困らないよう、ダッシュボードのクエリ側で webhook_id ごとに1行へ集約します（例：ROW_NUMBER() で最新1件に絞る）。ストリーミング挿入では直後の行を更新・削除しづらいことがあるため、すでに入ってしまった重複はこの方式で吸収するのが安全です

この2つを組み合わせると、「まず挿入時にほとんどを弾き、すり抜けた分は集計時に畳む」という形になり、どちらか一方だけよりも取りこぼしが減ります。なお、確実性をさらに高めたい場合は挿入前に同じキーが既にあるかを存在確認する方法もありますが、件数が多いと確認のクエリコストがかさむため、まずは上記の二段構えで十分なことが多いです。

小〜中規模であれば、row_ids での排除を入れたうえで「集計クエリ側でも畳む」ところから始めるのが運用しやすいです。webhook_id という一意キーさえ各行に残しておけば、後からどの方式にも移行できます。

⚠️ 注意点：署名検証は必須、再送への耐性も持たせる

公開URLでWebhookを受ける以上、次の2点は省略しないでください。

1. 署名検証は必ず行う

エンドポイントは誰でも叩けるURLです。署名検証を入れないと、偽の注文データを送り込まれてBigQueryのデータが汚染される恐れがあります。前述のHMAC検証は飾りではなく、この基盤の前提です。検証に失敗したリクエストは 401 で確実に拒否します。

2. 再送される前提で設計する

Webhookは「最低1回は届く」ものの「ちょうど1回」とは限りません。重複が来る前提で冪等に作るのが基本です。

加えて、処理が重くて応答が遅れると、Shopify側がタイムアウトと判断して再送を増やしてしまうことがあります。Cloud Functions の中では「署名検証 → 整形 → 挿入」だけを素早く行い、200 を早めに返すのが安全です。重い後処理が必要な場合は、いったん受領だけして別の処理へ渡す設計を検討してください。

なお、ShopifyのWebhookには配信に失敗し続けると自動で停止される仕組みがあります。エラーを放置するとある日Webhookが止まることもあるため、Cloud Functions のログとエラー通知は最初から設定しておくと安心です。Webhookの再送回数や停止条件などの仕様は変わる可能性があるので、最新の公式ドキュメントで確認してください。

まとめ

Shopifyの注文Webhookを Cloud Functions で受け、BigQueryへ流し込むことで、注文が入った数秒後には売上が分析基盤に反映される「速報レーン」を作れます。要点を振り返ります。

• 構成は Webhook → Cloud Functions → BigQuery の3点だけで、起動した分しか課金されない
• 公開URLで受けるため、HMAC署名検証は必須。生ボディに対して検証する
• Webhookは再送されるので、webhook_id を一意キーにして冪等に設計する
• 重複対策は、insert_rows_json の row_ids（insertId）で弾く＋集計クエリ側でも畳む二段構えが安全
• リアルタイム性を求めるなら insert_rows_json、件数が多くコストを抑えたいなら load 方式という選び方になる

この速報レーンは、日次バッチを置き換えるものではありません。普段は日次で十分なので、セールやキャンペーンなど「今すぐ知りたい」期間だけ併走させるのがコストと運用のバランスが良い使い方です。

そもそも注文データだけでなく、サイト側のユーザー行動とあわせて見たい場合は、計測タグの整備も欠かせません。Shopifyでのeコマース計測の基本は GTMでGA4のeコマース計測を完全設定する手順【Shopify対応】 で解説しているので、あわせて整えておくと、売上の速報と行動データの両面から当日の動きを追えるようになります。

今回紹介したコードはあくまで構成を示すための骨組みです。実際に組む際は、ShopifyとGoogle Cloudの最新の公式ドキュメントで現在の仕様を確認しながら進めてください。