GA4のデータとフォーム送信したデータをBigQueryに書き込み突合してみた

gcloud pubsub topics create form-submission-topic

bq mk --dataset --location=asia-northeast1 PROJECT_ID:ga4_analysis

bq mk --table PROJECT_ID:ga4_analysis.submissions \  user_id:STRING,client_id:STRING,form_type:STRING,name:STRING,email:STRING,message:STRING,submitted_at:TIMESTAMP

form-to-ga4-bq/
├── receiver/
│   ├── main.py
│   └── requirements.txt
├── ga4-sender/
│   ├── main.py
│   └── requirements.txt
└── bq-writer/
    ├── main.py
    └── requirements.txt

functions-framework==3.*
google-cloud-pubsub==2.*

import json
import uuid
import functions_framework
from google.cloud import pubsub_v1

PROJECT_ID = "PROJECT_ID"
TOPIC_ID = "form-submission-topic"

publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(PROJECT_ID, TOPIC_ID)

@functions_framework.http
def receive_form(request):
    """フォーム送信を受け取り、Pub/Subにパブリッシュする"""
    req = request.get_json(silent=True)
    if not req:
        return "リクエストボディが空です", 400

    # フォームの必須フィールド
    name = req.get("name")
    email = req.get("email")
    if not name or not email:
        return "nameとemailは必須です", 400

    # uuid4()で一意なuser_idを生成
    user_id = str(uuid.uuid4())

    # client_idもuuid4()で生成（ブラウザのclient_idと紐づけない場合）
    client_id = str(uuid.uuid4())

    # Pub/Subに送信するメッセージを構築
    message = {
        "user_id": user_id,
        "client_id": client_id,
        "form_type": req.get("form_type", "contact"),
        "name": name,
        "email": email,
        "message": req.get("message", ""),
        "debug_mode": req.get("debug_mode", 0),
    }

    # Pub/Subにパブリッシュ
    future = publisher.publish(topic_path, json.dumps(message).encode("utf-8"))
    message_id = future.result()

    print(f"Pub/Subにパブリッシュ完了: message_id={message_id}, user_id={user_id}")

    return {
        "status": "success",
        "user_id": user_id,
        "pubsub_message_id": message_id,
    }, 200

    user_id = str(uuid.uuid4())

    client_id = str(uuid.uuid4())

    future = publisher.publish(topic_path, json.dumps(message).encode("utf-8"))
    message_id = future.result()

functions-framework==3.*
requests==2.*

import base64
import json
import functions_framework
import requests

# GA4の設定
GA4_MEASUREMENT_ID = "測定ID"
GA4_API_SECRET = "APIシークレット"
MP_ENDPOINT = "https://www.google-analytics.com/mp/collect"

@functions_framework.cloud_event
def send_to_ga4(cloud_event):
    """Pub/Subメッセージを受け取り、GA4にMeasurement Protocolで送信する"""
    # Pub/Subメッセージのデコード
    message_data = base64.b64decode(cloud_event.data["message"]["data"]).decode("utf-8")
    submission = json.loads(message_data)

    user_id = submission["user_id"]
    client_id = submission["client_id"]

    # Measurement Protocolペイロードの構築
    event_params = {
        "engagement_time_msec": "10000",
        "form_type": submission.get("form_type", "contact"),
    }

    # debug_modeが指定されている場合はGA4のDebugViewに表示する
    if submission.get("debug_mode"):
        event_params["debug_mode"] = 1

    payload = {
        "client_id": client_id,
        "user_id": user_id,
        "events": [
            {
                "name": "form_submission",
                "params": event_params,
            }
        ],
    }

    url = f"{MP_ENDPOINT}?measurement_id={GA4_MEASUREMENT_ID}&api_secret={GA4_API_SECRET}"
    response = requests.post(url, json=payload)

    print(f"GA4送信完了: status={response.status_code}, user_id={user_id}")

@functions_framework.cloud_event
def send_to_ga4(cloud_event):

    message_data = base64.b64decode(cloud_event.data["message"]["data"]).decode("utf-8")
    submission = json.loads(message_data)

    event_params = {
        "engagement_time_msec": "10000",
        "form_type": submission.get("form_type", "contact"),
    }

    if submission.get("debug_mode"):
        event_params["debug_mode"] = 1

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

import base64
import json
import datetime
import functions_framework
from google.cloud import bigquery

# BigQueryの設定
BQ_PROJECT_ID = "PROJECT_ID"
BQ_DATASET = "ga4_analysis"
BQ_TABLE = "submissions"

bq_client = bigquery.Client()

@functions_framework.cloud_event
def write_to_bigquery(cloud_event):
    """Pub/Subメッセージを受け取り、BigQueryに書き込む"""
    # Pub/Subメッセージのデコード
    message_data = base64.b64decode(cloud_event.data["message"]["data"]).decode("utf-8")
    submission = json.loads(message_data)

    table_ref = f"{BQ_PROJECT_ID}.{BQ_DATASET}.{BQ_TABLE}"
    rows = [
        {
            "user_id": submission["user_id"],
            "client_id": submission["client_id"],
            "form_type": submission.get("form_type", "contact"),
            "name": submission["name"],
            "email": submission["email"],
            "message": submission.get("message", ""),
            "submitted_at": datetime.datetime.utcnow().isoformat(),
        }
    ]

    errors = bq_client.insert_rows_json(table_ref, rows)
    if errors:
        print(f"BigQuery書き込みエラー: {errors}")
        raise RuntimeError(f"BigQuery insert failed: {errors}")

    print(f"BigQuery書き込み完了: user_id={submission['user_id']}")

cd receiver

gcloud run deploy form-receiver \
    --source . \
    --function=receive_form \
    --base-image python312 \
    --region asia-northeast1 \
    --allow-unauthenticated

cd ga4-sender

gcloud run deploy ga4-sender \
    --source . \
    --function=send_to_ga4 \
    --base-image python312 \
    --region asia-northeast1 \
    --no-allow-unauthenticated

cd bq-writer

gcloud run deploy bq-writer \
    --source . \
    --function=write_to_bigquery \
    --base-image python312 \
    --region asia-northeast1 \
    --no-allow-unauthenticated

# サービスアカウントの作成
gcloud iam service-accounts create pubsub-invoker \
    --display-name="Pub/Sub Cloud Run Invoker"

# Cloud Runの呼び出し権限を付与
gcloud projects add-iam-policy-binding PROJECT_ID \
    --member="serviceAccount:pubsub-invoker@PROJECT_ID.iam.gserviceaccount.com" \
    --role="roles/run.invoker"

GA4_SENDER_URL=$(gcloud run services describe ga4-sender --region asia-northeast1 --format='value(status.url)')
BQ_WRITER_URL=$(gcloud run services describe bq-writer --region asia-northeast1 --format='value(status.url)')

# GA4送信用サブスクリプション
gcloud pubsub subscriptions create form-submission-ga4-sub \
    --topic=form-submission-topic \
    --push-endpoint=${GA4_SENDER_URL} \
    --push-auth-service-account=pubsub-invoker@PROJECT_ID.iam.gserviceaccount.com

# BigQuery書き込み用サブスクリプション
gcloud pubsub subscriptions create form-submission-bq-sub \
    --topic=form-submission-topic \
    --push-endpoint=${BQ_WRITER_URL} \
    --push-auth-service-account=pubsub-invoker@PROJECT_ID.iam.gserviceaccount.com

RECEIVER_URL=$(gcloud run services describe form-receiver --region asia-northeast1 --format='value(status.url)')

curl -X POST ${RECEIVER_URL} \
    -H "Content-Type: application/json" \
    -d '{
      "form_type": "contact",
      "name": "テスト太郎",
      "email": "test@example.com",
      "message": "お問い合わせテストです",
      "debug_mode": 1
    }'

{
  "status": "success",
  "user_id": "550e8400-e29b-41d4-a716-446655440000",
  "pubsub_message_id": "1234567890"
}

bq query --use_legacy_sql=false \
    "SELECT user_id, name, email, submitted_at FROM ga4_analysis.submissions ORDER BY submitted_at DESC LIMIT 5"

SELECT
    s.user_id,
    s.name,
    s.email,
    s.submitted_at,
    e.event_name,
    e.event_timestamp,
    e.device.category AS device_category,
    e.geo.country AS country,
    e.traffic_source.source AS traffic_source,
    e.traffic_source.medium AS traffic_medium
FROM
    `PROJECT_ID.ga4_analysis.submissions` AS s
LEFT JOIN
    `PROJECT_ID.analytics_PROPERTY_ID.events_*` AS e
ON
    s.user_id = e.user_id
WHERE
    _TABLE_SUFFIX BETWEEN
        FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY))
        AND FORMAT_DATE('%Y%m%d', CURRENT_DATE())
ORDER BY
    s.submitted_at DESC,
    e.event_timestamp ASC