C データエンジニアリング — パーソナライズ導線A/BテストCVR集計基盤（1ユーザー1行のab_assignmentsと1ユーザーN行のconversion_eventsを事前集約せず直接JOINしfan-out重複でCVRが水増し × ダッシュボードクエリのCURRENT_TIMESTAMP()で結果キャッシュが一切効かない × 割当比率50:50の閾値がSQLベタ書きでdbt Unit Testsが無くデプロイ時の書き間違いを検知できず × user_idクラスタリング未設定 × dbt incrementalがappendでリトライ時の重複が伝播 × Argo Workflowsのバッチ実行がダッシュボード閲覧ピークと重なりスロット輻輳 × 統計的有意性検定が無く目視判断）（MOps 販促システムチーム「新ロジックが有意に優れている」誤判断寸前 Bad→Good）

概要

問題

ECサイト MOps 販促システムチームでは、トップページのレコメンド導線を2パターン（control: 従来ロジック / variant: 新ロジック）でA/Bテストしている。ユーザーには初回訪問時にMOD(FARM_FINGERPRINT(user_id), 100) < 50というハッシュベースの割当ロジックで50:50の比率でcontrol/variantを割り当て、raw_events.ab_assignments（1ユーザー1行）に記録する。ユーザーが商品を購入するたびにraw_events.conversion_events（1ユーザーにつき複数回発生しうる）にイベントが追記される。この2つの生ログをdbt Core 1.9+でmart_ab_test_conversion（variant別CVR集計マート。Looker Studioの「A/Bテスト結果ダッシュボード」から5分間隔で参照される）に変換し、Argo Workflows CronWorkflowで毎時実行している。

先週起きた出来事: ダッシュボード上でvariantのCVRがcontrolを大きく上回り、担当者が「新ロジックの方が統計的に有意に優れている」と判断しかけたが、社内レビューで疑義が入り原因調査したところ、以下の実装には7つの設計・データ品質・コスト上の問題が潜んでいることが判明した。

問題点を全て洗い出し、ウィンドウ関数/事前集約によるJOIN fan-out重複の解消 × クエリ結果キャッシュを効かせる決定的なクエリ設計 × dbt Unit Tests（dbt Core 1.9+）による割当ロジックの回帰防止 × user_idクラスタリング × dbt incremental merge戦略 × BigQuery Reservation API（baseline + autoscale） × 二項比率検定（z検定）による統計的有意性判定を活用したBad→Goodリファクタリングを行ってください。

制約・前提条件

• BigQuery Standard SQL、dbt Core 1.9+（incremental_strategy: merge、Unit Testsがいずれも利用可能）、Argo Workflows 3.5+ CronWorkflow
• raw_events.ab_assignments: user_id（1ユーザー1行）, variant, assigned_at
• raw_events.conversion_events: user_id, conversion_id, order_amount, converted_at（1ユーザーにつき0回以上、複数回発生しうる。セール期間中は同一ユーザーが複数回購入することも珍しくない）
• 現状のmart_ab_test_conversionは両テーブルをuser_idで直接JOINしており、複数回コンバージョンしたユーザーの割当行がコンバージョン件数分だけ複製される。GROUP BY variantでユーザー数をカウントすると、複数回購入したユーザーが多い方のvariantで「コンバージョンしたユーザー数」自体が水増しされる
• Looker Studioが発行するダッシュボードクエリのWHERE句にWHERE assigned_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)という非決定的関数CURRENT_TIMESTAMP()が含まれ、BigQueryの結果キャッシュ（同一クエリ文字列・対象テーブル未更新なら24時間有効）が一切効かず、5分間隔の自動リフレッシュのたびに対象データをフルスキャンしている
• 割当比率50:50を決めるMOD(FARM_FINGERPRINT(user_id), 100) < 50はdbtモデル内にCASE文でベタ書きされ、閾値50という数値にdbt Unit Testsによる検証が無い。今回のインシデントの真因のひとつは、直近のデプロイで担当者が誤って閾値を< 57と書き間違えたことであり、variant側に本来より多くのユーザーが割り当てられていた
• ab_assignments・conversion_eventsいずれもuser_idでのJOINが主要アクセスパターンだが、クラスタリングが設定されておらず、数千万行規模のテーブル同士のJOINで不要なシャッフルコストが発生している
• mart_ab_test_conversionを作るdbt incremental modelはincremental_strategy: append（追記のみ）になっており、Argo Workflowsのリトライ（ジョブがタイムアウトして再実行されるケースがある）で同一ユーザーの割当行が複数回insertされると、そのまま重複がマートに伝播する
• Argo Workflows CronWorkflowはdbt実行を毎時0分に固定でスケジュールしており、これがLooker Studio利用者（経営陣・現場責任者）が集中する朝会・夕会の時間帯（9時・18時台）のダッシュボード閲覧ピークと重なる。BigQueryはオンデマンドスロットのみで運用しており、バッチクエリとダッシュボードの対話的クエリがスロットを奪い合い、表示が数十秒単位で遅延する
• CVRの差をvariant_cvr - control_cvrの単純な割合比較でしか可視化しておらず、サンプルサイズや信頼区間、二項比率検定（z検定）などの統計的有意性の判定ロジックが組み込まれていない

悪いコード (Before)

📋 コピー-- config(materialized='incremental')  ※incremental_strategy未指定=append相当
-- 問題①: conversion_eventsを事前集約せずuser_idで直接JOIN → fan-out
-- 問題③: 割当閾値50がCASE文にベタ書き、Unit Testsが無い
-- 問題④: クラスタリング未設定
-- 問題⑤: incremental_strategyがappendでリトライ時の重複が伝播
SELECT
  a.user_id,
  CASE WHEN MOD(FARM_FINGERPRINT(a.user_id), 100) < 50
       THEN 'control' ELSE 'variant' END AS variant,
  a.assigned_at,
  c.conversion_id IS NOT NULL AS is_converted  -- 複数回購入で行が複製される
FROM {{ ref('stg_ab_assignments') }} a
LEFT JOIN {{ ref('stg_conversion_events') }} c USING (user_id)

📋 コピー-- 問題②: CURRENT_TIMESTAMP()を含むため結果キャッシュ(24h)が一切効かない
-- 5分間隔の自動リフレッシュのたびにフルスキャンが発生
SELECT variant, COUNT(*) AS n, COUNTIF(is_converted) AS conversions
FROM `mops-project.marts.mart_ab_test_conversion`
WHERE assigned_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
GROUP BY variant
-- 問題⑦: この後、割合比較のみで「有意」と判断されており
--        サンプルサイズ・信頼区間・有意性検定のロジックが無い

ヒント（段階的開示）

-- dbtモデルの骨格（事前集約してからJOIN）
WITH conversion_agg AS (
  SELECT user_id, COUNT(*) > 0 AS is_converted
  FROM {{ ref('stg_conversion_events') }}
  GROUP BY user_id                      -- fan-outを防ぐ要
)
SELECT a.user_id, a.variant, COALESCE(c.is_converted, FALSE) AS is_converted
FROM {{ ref('stg_ab_assignments') }} a
LEFT JOIN conversion_agg c USING (user_id)  -- 1:1のJOINになる

-- 二項比率検定(z検定)の骨格
WITH stats AS (
  SELECT variant, COUNT(*) AS n, SUM(CAST(is_converted AS INT64)) AS conversions
  FROM {{ ref('mart_ab_test_conversion') }}
  GROUP BY variant
)
-- pooled proportion → 標準誤差 → z値 の順で計算する

問題点分析（7点）

パイプライン比較図 — Bad vs Good の比較

模範解答

📋 コピー-- config(materialized='incremental', incremental_strategy='merge', unique_key='user_id')
--
-- 修正①: conversion_eventsをuser_id単位に事前集約してからJOINし、fan-outを防ぐ
-- 修正⑤: incremental_strategy=mergeにより、Argo Workflowsのリトライで
--        割当行が複数回insertされても、user_id単位で最終的に1行へ収束する

WITH assignments AS (
  SELECT user_id, variant, assigned_at
  FROM {{ ref('stg_ab_assignments') }}
  {% if is_incremental() %}
  WHERE assigned_at > (SELECT MAX(assigned_at) FROM {{ this }})
  {% endif %}
),

-- 修正①: 1ユーザーにつき複数回のコンバージョンを、ここで1行に集約してから
--        assignmentsとJOINする。JOIN後もuser_id単位の粒度が保たれる。
conversion_agg AS (
  SELECT
    user_id,
    COUNT(*) > 0 AS is_converted,
    SUM(order_amount) AS total_conversion_amount,
    MIN(converted_at) AS first_converted_at
  FROM {{ ref('stg_conversion_events') }}
  GROUP BY user_id
)

SELECT
  a.user_id,
  a.variant,
  a.assigned_at,
  COALESCE(c.is_converted, FALSE) AS is_converted,
  COALESCE(c.total_conversion_amount, 0) AS total_conversion_amount,
  c.first_converted_at
FROM assignments a
LEFT JOIN conversion_agg c USING (user_id)  -- 修正①: 1:1のJOINになりfan-outしない

📋 コピーversion: 2

models:
  - name: mart_ab_test_conversion
    config:
      materialized: incremental
      incremental_strategy: merge   # 修正⑤
      unique_key: user_id           # 修正⑤: リトライ時の重複をここで収束させる
    columns:
      - name: user_id
        data_type: string
        constraints: [{ type: not_null }, { type: primary_key }]
      - name: variant
        data_type: string
        constraints: [{ type: not_null }, { type: check, expression: "variant IN ('control', 'variant')" }]

vars:
  ab_test_split_threshold: 50   # 修正③: 割当比率をdbt変数として一元管理

# 修正③: 割当ロジックの閾値が意図通りであることを固定入力で検証する
unit_tests:
  - name: test_assignment_split_is_50_50
    description: >
      MOD(FARM_FINGERPRINT(user_id), 100) < {{ var('ab_test_split_threshold') }} という
      割当ロジックが、閾値50のとき固定ハッシュ値の集合に対して意図通り約50%を
      variantに割り当てることを検証する。閾値の書き間違い(50→57)をCIで検知するための回帰テスト。
    model: stg_ab_assignments
    given:
      - input: ref('raw_ab_hash_fixture')
        rows:
          - { user_id: "u1", hash_mod_100: 10 }   # < 50 → control
          - { user_id: "u2", hash_mod_100: 49 }   # < 50 → control
          - { user_id: "u3", hash_mod_100: 50 }   # >= 50 → variant
          - { user_id: "u4", hash_mod_100: 99 }   # >= 50 → variant
    expect:
      rows:
        - { user_id: "u1", variant: "control" }
        - { user_id: "u2", variant: "control" }
        - { user_id: "u3", variant: "variant" }
        - { user_id: "u4", variant: "variant" }

📋 コピーCREATE TABLE `mops-project.raw_events.ab_assignments`
(
  user_id STRING NOT NULL,
  variant STRING NOT NULL,
  assigned_at TIMESTAMP NOT NULL
)
PARTITION BY DATE(assigned_at);
-- 問題④: CLUSTER BYが無くuser_idでのJOINが重い

📋 コピー-- 修正④: user_idでのJOINが主要アクセスパターンのためクラスタリングを設定
CREATE TABLE IF NOT EXISTS `mops-project.raw_events.ab_assignments`
(
  user_id STRING NOT NULL,
  variant STRING NOT NULL,
  assigned_at TIMESTAMP NOT NULL
)
PARTITION BY DATE(assigned_at)
CLUSTER BY user_id;   -- 修正④

CREATE TABLE IF NOT EXISTS `mops-project.raw_events.conversion_events`
(
  conversion_id STRING NOT NULL,
  user_id STRING NOT NULL,
  order_amount NUMERIC,
  converted_at TIMESTAMP NOT NULL
)
PARTITION BY DATE(converted_at)
CLUSTER BY user_id;   -- 修正④

📋 コピーSELECT variant, COUNT(*) AS n, COUNTIF(is_converted) AS conversions
FROM `mops-project.marts.mart_ab_test_conversion`
WHERE assigned_at >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
GROUP BY variant
-- 問題②: 秒単位で値が変わる関数のため結果キャッシュが一切効かない

📋 コピー-- 修正②: CURRENT_DATE()は1日1回しか値が変わらないため
--        同日中に発行される同一クエリは結果キャッシュ(24時間)がヒットする
SELECT
  variant,
  COUNT(*) AS n,
  COUNTIF(is_converted) AS conversions,
  SAFE_DIVIDE(COUNTIF(is_converted), COUNT(*)) AS cvr
FROM `mops-project.marts.mart_ab_test_conversion`
WHERE DATE(assigned_at) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)
GROUP BY variant

📋 コピー# 修正⑥: ダッシュボード用とバッチ用でスロットプールを分離し、
#        オンデマンドスロットの奪い合いを構造的に避ける
resource "google_bigquery_reservation" "ab_test_batch" {
  name     = "ab-test-batch-reservation"
  project  = "mops-project"
  location = "asia-northeast1"

  slot_capacity = 100          # baseline: 常時確保するスロット
  edition       = "ENTERPRISE"

  autoscale {
    max_slots = 300            # ピーク時(dbt実行×ダッシュボード集中)は自動拡張
  }
}

resource "google_bigquery_reservation_assignment" "batch_assignment" {
  reservation = google_bigquery_reservation.ab_test_batch.id
  job_type    = "PIPELINE"     # dbt実行(バッチ)をこのReservationに割り当て
  project     = "mops-project"
}

📋 コピーapiVersion: argoproj.io/v1alpha1
kind: CronWorkflow
metadata:
  name: ab-test-mart-hourly
spec:
  # 修正⑥: 朝会(9時台)・夕会(18時台)のダッシュボード閲覧ピークを避け、
  #        毎時25分・55分（ピーク直前を避けた分）に実行をずらす
  schedules:
    - "25,55 0-8,10-17,19-23 * * *"   # 9時台・18時台はスキップ
  concurrencyPolicy: Forbid
  workflowSpec:
    entrypoint: dbt-run
    templates:
      - name: dbt-run
        container:
          image: gcr.io/mops-project/dbt:1.9
          command: ["dbt", "build", "--select", "mart_ab_test_conversion"]

📋 コピー-- 二項比率のz検定(正規近似)。有意水準5%(|z| >= 1.96)を基準に判定する。
WITH stats AS (
  SELECT
    variant,
    COUNT(*) AS n,
    COUNTIF(is_converted) AS conversions,
    SAFE_DIVIDE(COUNTIF(is_converted), COUNT(*)) AS p
  FROM `mops-project.marts.mart_ab_test_conversion`
  GROUP BY variant
),
pivoted AS (
  SELECT
    MAX(IF(variant = 'control', n, NULL)) AS n_control,
    MAX(IF(variant = 'control', p, NULL)) AS p_control,
    MAX(IF(variant = 'variant', n, NULL)) AS n_variant,
    MAX(IF(variant = 'variant', p, NULL)) AS p_variant
  FROM stats
),
pooled AS (
  SELECT
    *,
    SAFE_DIVIDE(p_control * n_control + p_variant * n_variant, n_control + n_variant) AS p_pool
  FROM pivoted
)
SELECT
  n_control, p_control, n_variant, p_variant,
  p_variant - p_control AS diff,
  SAFE_DIVIDE(
    p_variant - p_control,
    SQRT(p_pool * (1 - p_pool) * (SAFE_DIVIDE(1, n_control) + SAFE_DIVIDE(1, n_variant)))
  ) AS z_score,
  ABS(SAFE_DIVIDE(
    p_variant - p_control,
    SQRT(p_pool * (1 - p_pool) * (SAFE_DIVIDE(1, n_control) + SAFE_DIVIDE(1, n_variant)))
  )) >= 1.96 AS is_significant_at_5pct
FROM pooled;

実行例（input → output）

input（改善前のシナリオ）:

7日間でcontrolに49,800ユーザー、variant（閾値の書き間違いにより実際は本来より多い）に52,300ユーザーが割り当てられた。うちvariant側でセール期間中に3回購入したユーザーが2,000人おり、conversion_eventsとのJOINでこの2,000人の割当行が3行ずつに複製された。

output（Before・問題のある実装）:

JOIN後の行数がユーザー数と一致せず、variant側は複製された分だけ「コンバージョンしたユーザー数」が多く見え、control CVR=8.1% に対し variant CVR=11.4% と表示される。ダッシュボードはCURRENT_TIMESTAMP()のせいで5分ごとにフルスキャンし直され、月間BigQueryコストも増大している。単純な割合比較のみで「variantが有意に優れている」と判断されかけた。

output（After・改善後）:

conversion_aggでユーザー単位に事前集約してからJOINするため、複数回購入したユーザーも1行のまま扱われ、CVRが正しい母数で計算される。割当閾値のUnit Testsがdbt build時点で書き間違いを検知するため、正しい50:50の母集団で比較できる。二項比率のz検定を実施した結果|z| = 1.42（有意水準5%の基準1.96未満）となり、実際には統計的有意差は無かったことが判明する。ダッシュボードクエリは決定的な日付条件に変わり結果キャッシュがヒットするため、5分間隔のリフレッシュでも同日内はスキャンコストが発生しない。

ポイント解説

実務への応用

MOpsチームが運用するA/Bテスト集計基盤は、レコメンド導線に限らずクーポン施策・バナー配置・価格表示のテストなど横展開が多い場面であり、「割当テーブル(1:1) × イベントテーブル(1:N)」という同型の構造を持つケースが繰り返し登場する。今回のJOIN fan-out対策（事前集約してからJOIN）は、A/Bテスト集計に限らず「ユーザー単位の指標をイベントログから作る」あらゆるマート設計の基本パターンとして再利用できる。また、クエリキャッシュを意識したダッシュボードクエリ設計は、A/Bテストに限らず全社ダッシュボードのコスト最適化にそのまま応用できる実務的なチェック項目になる。

今日のまとめ

次のステップ

• 発展問題: 今回はvariantが2値（control/variant）だが、3パターン以上の多変量テスト（A/B/n テスト）に拡張する場合、二項比率検定ではなくカイ二乗検定（またはBonferroni補正付きの多重比較）が必要になる。mart_ab_test_conversionをn変量対応に拡張し、多重比較補正込みの有意性判定SQLを設計せよ。
• 参考: BigQueryクエリ結果キャッシュの条件（決定的クエリ） / dbt Unit Tests（dbt Core 1.9+） / dbt incremental merge strategy / BigQuery Reservation API（baseline + autoscale） / 二項比率のz検定・多重比較補正

自己評価（あとで記入）