C データエンジニアリング — dbt incremental + insert_overwrite × BigQuery パーティション剪定 × COALESCE × SAFE_DIVIDE × Unit Tests（MOps BQ コスト97%削減 Bad→Good）

📋 コピー-- models/mart/mart_daily_revenue.sql
-- 日次売上マート: SKU×日付の集計
{{ config(
    materialized='incremental',              -- 修正①: テーブル全リビルドを廃止
    incremental_strategy='insert_overwrite',
    partition_by={
        'field': 'order_date',
        'data_type': 'date',
        'granularity': 'day'
    },
    cluster_by=['sku_id']
) }}

{% set LOOKBACK_DAYS = 3 %}

SELECT
    order_date,
    sku_id,
    SUM(revenue)                              AS total_revenue,
    COUNT(DISTINCT user_id)                   AS unique_buyers,
    SUM(quantity)                             AS total_qty,
    -- 修正④: SAFE_DIVIDE でゼロ除算を NULL に変換
    -- （購入者ゼロのセグメントでもクエリが落ちない）
    SAFE_DIVIDE(
        SUM(revenue),
        NULLIF(COUNT(DISTINCT user_id), 0)
    )                                         AS arpu
FROM {{ ref('stg_orders') }}
{% if is_incremental() %}
WHERE order_date >= DATE_SUB(CURRENT_DATE(), INTERVAL {{ LOOKBACK_DAYS }} DAY)
{% endif %}
GROUP BY 1, 2

📋 コピー# models/mart/schema.yml
# dbt Core 1.8+ Unit Tests 定義
version: 2

models:
  - name: mart_daily_revenue
    description: "日次売上マート（SKU × 日付）"
    columns:
      - name: order_date
        tests: [not_null]
      - name: sku_id
        tests: [not_null]
      - name: total_revenue
        tests: [not_null]
      - name: arpu
        description: "購入者がいない場合は NULL（SAFE_DIVIDE）"

unit_tests:
  # テスト①: 購入者ゼロの場合に arpu が NULL になること（修正④保証）
  - name: test_arpu_null_when_no_buyers
    model: mart_daily_revenue
    given:
      - input: ref('stg_orders')
        rows:
          - {order_date: "2026-06-01", sku_id: "SKU-001",
             revenue: 0.0, user_id: null, quantity: 0}
    expect:
      rows:
        - {order_date: "2026-06-01", sku_id: "SKU-001", arpu: null}

  # テスト②: revenue が正常に集計されること
  - name: test_revenue_aggregation
    model: mart_daily_revenue
    given:
      - input: ref('stg_orders')
        rows:
          - {order_date: "2026-06-01", sku_id: "SKU-002",
             revenue: 1000.0, user_id: "U1", quantity: 2}
          - {order_date: "2026-06-01", sku_id: "SKU-002",
             revenue: 500.0,  user_id: "U2", quantity: 1}
    expect:
      rows:
        - {order_date: "2026-06-01", sku_id: "SKU-002",
           total_revenue: 1500.0, unique_buyers: 2,
           total_qty: 3, arpu: 750.0}

  # テスト③: NULL quantity/unit_price が revenue=0 になること（修正②保証）
  - name: test_revenue_non_negative_on_null_input
    model: stg_orders
    given:
      - input: source('raw', 'orders')
        rows:
          - {order_id: "O1", user_id: "U1", sku_id: "SKU-003",
             quantity: null, unit_price: 500.0,
             status: "purchased", created_at: "2026-06-01 10:00:00 UTC"}
    expect:
      rows:
        - {order_id: "O1", revenue: 0.0}

📋 コピー"""BigQuery インクリメンタルローダー。

Argo Workflows から毎時呼び出される。
直近 LOOKBACK_DAYS 日のデータのみを raw_orders に INSERT し、
dbt incremental run + unit tests を実行する。
"""
from __future__ import annotations

import argparse
import subprocess
import sys
from dataclasses import dataclass

from google.cloud import bigquery


# ---- 定数 ----------------------------------------------------------------
LOOKBACK_DAYS: int = 3          # 遅延データの最大日数（修正⑤⑥と合わせる）
DBT_SELECT_MODELS: str = "stg_orders mart_daily_revenue"
# -------------------------------------------------------------------------


@dataclass(frozen=True, slots=True)
class LoaderConfig:
    """ローダー設定値オブジェクト。

    Attributes:
        project_id: GCP プロジェクト ID
        dataset: BigQuery データセット名
        lookback_days: 処理対象の日数（遅延データ考慮）
    """
    project_id: str
    dataset: str
    lookback_days: int = LOOKBACK_DAYS


def load_orders_incremental(config: LoaderConfig) -> None:
    """直近 lookback_days のオーダーデータを raw_orders にインクリメンタルロードする。

    Args:
        config: ローダー設定

    Raises:
        google.api_core.exceptions.GoogleAPIError: BQ クエリ失敗時
    """
    client = bigquery.Client(project=config.project_id)

    # 修正⑤: パーティション剪定を有効化（直近3日のみ INSERT）
    # DATE(created_at) で order_date パーティションを剪定する
    query = f"""
        INSERT INTO `{config.project_id}.{config.dataset}.raw_orders`
        SELECT *
        FROM `{config.project_id}.{config.dataset}.orders_staging`
        WHERE
            status IN ('purchased', 'shipped')
            AND DATE(created_at) >= DATE_SUB(
                CURRENT_DATE(), INTERVAL {config.lookback_days} DAY
            )
    """
    job = client.query(query)
    job.result()  # 完了待機（例外は呼び出し元に伝播）
    print(f"[INFO] BQ INSERT 完了: {job.num_dml_affected_rows} 行")


def run_dbt_incremental(select: str = DBT_SELECT_MODELS) -> None:
    """dbt incremental run + unit tests を実行する。

    --full-refresh は渡さない（修正⑥）。
    テスト失敗時はゼロ以外の終了コードで sys.exit する。

    Args:
        select: dbt --select に渡すモデル名（スペース区切り）
    """
    # 修正⑥: --full-refresh を削除し通常の incremental run を実行
    run_result = subprocess.run(
        ["dbt", "run", "--select", select],
        capture_output=True, text=True,
    )
    if run_result.returncode != 0:
        print(f"[ERROR] dbt run 失敗:\n{run_result.stderr}", file=sys.stderr)
        sys.exit(run_result.returncode)

    # Unit Tests は dbt test で別途実行（dbt 1.8+ unit_tests ブロック）
    test_result = subprocess.run(
        ["dbt", "test", "--select", select],
        capture_output=True, text=True,
    )
    if test_result.returncode != 0:
        print(f"[ERROR] dbt test 失敗:\n{test_result.stderr}", file=sys.stderr)
        sys.exit(test_result.returncode)

    print("[INFO] dbt run + test 完了")


def main() -> None:
    """エントリポイント。"""
    parser = argparse.ArgumentParser(description="BQ インクリメンタルローダー")
    parser.add_argument("--project", required=True)
    parser.add_argument("--dataset", required=True)
    parser.add_argument("--lookback-days", type=int, default=LOOKBACK_DAYS)
    args = parser.parse_args()

    config = LoaderConfig(
        project_id=args.project,
        dataset=args.dataset,
        lookback_days=args.lookback_days,
    )
    load_orders_incremental(config)
    run_dbt_incremental()


if __name__ == "__main__":
    main()