C データエンジニアリング — BigQuery ROW_NUMBER ラスト・タッチ attribution × dbt source freshness × PIVOT 演算子 × SELECT * スキャン削減 × N+1 Python ループ廃止（MOps キャンペーン attribution Bad→Good）

sources.yml — freshness 設定（修正⑥）

📋 コピー# models/sources.yml
version: 2

sources:
  - name: events
    database: my-gcp-project
    schema: events
    tables:
      - name: email_clicks
        description: "メールリンクのクリックイベント（Pub/Sub → BQ streaming insert）"
        # 修正⑥: freshness 監視 — Pub/Sub → BQ パイプライン遅延を自動検知
        loaded_at_field: _ingested_at   # streaming insert のシステム列
        freshness:
          warn_after:  {count: 2, period: hour}   # 2時間以上新規データなし → warn
          error_after: {count: 6, period: hour}   # 6時間以上 → error (アラート発火)
        columns:
          - name: campaign_id
            tests: [not_null]
          - name: user_id
            tests: [not_null]
          - name: event_date
            tests: [not_null]

  - name: orders
    database: my-gcp-project
    schema: orders
    tables:
      - name: order_items
        description: "注文明細（Cloud Spanner → BQ レプリケーション）"
        loaded_at_field: _synced_at
        freshness:
          warn_after:  {count: 1, period: hour}
          error_after: {count: 3, period: hour}
        columns:
          - name: order_id
            tests: [not_null, unique]
          - name: user_id
            tests: [not_null]
          - name: revenue
            tests:
              - not_null
              - dbt_utils.accepted_range:
                  min_value: 0

DataDog freshness アラート統合（Argo Workflows）

📋 コピー# freshness_alert.py — dbt source freshness を DataDog メトリクスに送信
"""dbt source freshness を DataDog statsd メトリクスに変換して送信。

CronWorkflow: 毎時 dbt source freshness --output json → このスクリプトに渡す
"""
from __future__ import annotations
import json, os, sys
from datadog import initialize, statsd   # pip install datadog

initialize(statsd_host="localhost", statsd_port=8125)

def report_freshness(json_output: str) -> None:
    """dbt source freshness の JSON 出力を DataDog ゲージメトリクスに変換。

    Args:
        json_output: `dbt source freshness --output json` の標準出力

    Returns:
        None（DataDog に statsd.gauge を送信）
    """
    results = json.loads(json_output)
    for result in results.get("results", []):
        source_name: str = result["unique_id"]   # e.g. "source.my_project.events.email_clicks"
        status: str      = result["status"]      # "pass" | "warn" | "error"
        elapsed_hours: float = result.get("max_loaded_at_time_ago_in_s", 0) / 3600

        # DataDog メトリクス: dbt.source.freshness.elapsed_hours
        statsd.gauge(
            "dbt.source.freshness.elapsed_hours",
            elapsed_hours,
            tags=[
                f"source:{source_name}",
                f"status:{status}",
                f"env:{os.environ.get('DD_ENV', 'production')}",
            ],
        )

        if status in ("warn", "error"):
            statsd.event(
                title=f"dbt source freshness {status.upper()}: {source_name}",
                message=f"経過時間: {elapsed_hours:.1f}h — パイプライン遅延を確認してください",
                alert_type=status,
                tags=[f"source:{source_name}"],
            )

if __name__ == "__main__":
    report_freshness(sys.stdin.read())

mart_campaign_attribution_pivot.sql — BigQuery PIVOT（修正⑦）

📋 コピー-- models/mart/mart_campaign_attribution_pivot.sql
-- 修正⑦: pandas pivot_table → BigQuery PIVOT 演算子（Pod メモリ消費ゼロ）
-- キャンペーン別・月別の売上貢献を横持ち変換（直近6ヶ月）
{{
  config(
    materialized = 'table',   -- PIVOT 結果は table で完全リビルド
  )
}}

SELECT *
FROM (
  SELECT
    campaign_id,
    FORMAT_DATE('%Y-%m', click_date) AS ym,    -- 例: '2026-01'
    attributed_revenue
  FROM {{ ref('mart_campaign_attribution') }}
  WHERE click_date >= DATE_SUB(CURRENT_DATE('Asia/Tokyo'), INTERVAL 6 MONTH)
)
PIVOT (
  -- 修正⑦: BigQuery PIVOT 演算子で BQ クラスター上処理 → OOM なし
  -- 注意: dbt では静的 PIVOT のみサポート（動的 PIVOT は EXECUTE IMMEDIATE を使う）
  SUM(attributed_revenue) AS revenue
  FOR ym IN (
    '2025-12',
    '2026-01',
    '2026-02',
    '2026-03',
    '2026-04',
    '2026-05',
    '2026-06'
  )
)

動的 PIVOT（Jinja マクロ版）

📋 コピー-- macros/generate_pivot_months.sql
-- dbt Jinja で直近6ヶ月の月リストを動的生成
{% macro generate_pivot_months(n=6) %}
  {% set months = [] %}
  {% for i in range(n, 0, -1) %}
    {% set _ = months.append(
      "'" ~ (modules.datetime.date.today().replace(day=1) - modules.datetime.timedelta(days=30*i)).strftime('%Y-%m') ~ "'"
    ) %}
  {% endfor %}
  {{ months | join(', ') }}
{% endmacro %}

-- 使用例（mart_campaign_attribution_pivot.sql で呼び出し）:
-- FOR ym IN ( {{ generate_pivot_months(6) }} )

dbt_project.yml — 変数定義（修正③）

📋 コピー# dbt_project.yml（抜粋）
name: mops_dbt
version: '1.0.0'
config-version: 2

# 修正③: マジックナンバーを vars で一元管理
vars:
  attribution_window_days: 7   # クリック後 N 日以内の注文を貢献とみなす

models:
  mops_dbt:
    staging:
      +materialized: view
    mart:
      +materialized: incremental
      +partition_by:
        field: click_date
        data_type: date
      +cluster_by: [campaign_id]

問題点と改善方法の対応表