C データエンジニアリング — 遅延到着と重複のある日次 incremental を直す（水位は届いた時刻、影響日を丸ごと再計算）

概要

問題

MOps 販促システムチームの mart_campaign_daily（Pub/Sub → BigQuery → dbt 1.9 incremental、毎晩 02:00 JST）の日別値引き額が、財務の集計と翌週になっても合わない。

📋 コピーselect date(event_ts) as event_date, count(*) as redemptions,
       sum(discount_yen) as discount_yen, max(event_ts) as max_event_ts
from {{ ref('stg_coupon_redemptions') }}      -- 重複除去なしの view
{% if is_incremental() %}
where event_ts > (select max(max_event_ts) from {{ this }})
{% endif %}
group by 1        -- config: merge, unique_key='event_date'。テストは not_null / unique のみ

• アプリは圏外でイベントを溜めて後から送る。Pub/Sub は at-least-once。日付は JST の営業日で数える
• 履歴 約 1.5 TiB・1日 3.8 GiB 仮定。$6.25/TiB 未確認。財務は翌営業日確定を希望

期待する回答形式: 問題点 7 点／改善後コードと照合テスト／実行例／適用パターン／迷った案と前提が変わったときの判断／扱わなかった範囲

ヒント（段階的開示）

touched = distinct jst_date(event_ts) where loaded_at in (watermark, T] and date >= today - L
touched の各日を raw から再計算（event_id で重複除去）→ partition 単位で上書き
照合: 直近 14 日の mart と raw の再集計を日別に比較 / 週次: 直近 W 日を全再計算

問題点分析（7点）

誤差の比較（SVG）

各行は Good から 1 点だけ Bad に戻した誤差（Bad だけは全部 Bad）。√ で描いているため棒の長さの比は誤差の比ではない。

Bad → Good（コード差分）

📋 コピー# Bad: 起きた時刻の水位・UTC・足りない分を上書き
select date(event_ts) as event_date, count(*), sum(discount_yen),
       max(event_ts) as max_event_ts
from stg_coupon_redemptions      -- 重複あり
where event_ts > (select max(max_event_ts) from {{ this }})
group by 1                       -- merge: 新規分だけで日合計を上書き

📋 コピー# Good: 届いた時刻の水位・影響日の再計算・JST・重複除去
touched = distinct date(event_ts,'Asia/Tokyo')
  where loaded_at > watermark         -- 届いた行が触った日
    and date >= today - L             -- 直近 L 日だけ
select date(event_ts,'Asia/Tokyo'), count(*), sum(discount_yen)
from (row_number() over (partition by event_id ...) = 1) -- 重複除去
where date in touched               -- insert_overwrite で置き換え

模範解答

dbt モデル（BigQuery・未実行のスケッチ）

📋 コピー{{ config(materialized='incremental',
    incremental_strategy='insert_overwrite',   # 影響日を partition 単位で置き換える
    partition_by={'field': 'event_date_jst', 'data_type': 'date'}) }}
{% set lookback_days = var('lookback_days', 3) %} # L: 遅延の p99 と費用で決める
{% set safety_minutes = 10 %}                  # commit 順の揺れへの余白（未再現）
with dedup as (
  select * except(rn) from (
    select *, row_number() over (partition by event_id order by loaded_at) as rn -- 再配信は 1 件
    from {{ source('app','coupon_redemptions') }}
    where date(event_ts,'Asia/Tokyo') in unnest([{{ touched | join(',') }}]) -- 窓を明示
  ) where rn = 1)
select date(event_ts,'Asia/Tokyo') as event_date_jst, count(*) as redemptions,
       sum(discount_yen) as discount_yen, max(loaded_at) as loaded_watermark,
       date(event_ts,'Asia/Tokyo') <= date_sub(current_date('Asia/Tokyo'), interval 14 day) as is_final
from dedup group by 1

照合テスト（直近 14 日の mart と raw の再集計）

📋 コピー-- severity: warn → 週次の再計算を起動。未実行
select t.d, t.n, m.redemptions, t.y, m.discount_yen
from truth t left join {{ ref('mart_campaign_daily') }} m on m.event_date_jst = t.d
where m.redemptions is distinct from t.n or m.discount_yen is distinct from t.y

シミュレータ（実測の元・実行済み）

📋 コピーnew_w = f"loaded_at <= {T} AND loaded_at > {last_t}"   # 水位は届いた時刻
targets = {d for d in new if d >= run_date - L}         # 触った日のうち直近 L 日
# 各日を丸ごと再計算（event_id で GROUP BY して重複除去）→ mart[d] = 置き換え

実行例

📋 コピーraw rows 608879  unique events 600000
lag(hours) first-load  p50 0.01  p90 0.06  p99 34.32  p99.9 135.53   # 実測・合成
S0 Bad                              19/19 日不一致  -84.66%   読んだ行 590,051
S6 Good 日次(L=3)                    19/19          -0.44%   2,417,246
S6 + 週次 14日窓                      7/19          -0.01%   3,431,523
S6 + 週次 全期間                      0/19           0.00%   4,811,578
L 掃引: 1 -2.90% / 2 -0.98% / 3 -0.44% / 5 -0.13% / 7 -0.08% / 14 0.00%
$ python3 test_sim.py
Ran 6 tests ... OK   # 遅延・重複・JST境界・部分上書き・照合・窓外 の 6 件

仮定: 履歴 1.5 TiB・1日 3.8 GiB。$6.25/TiB は未確認。費用の削減は副次で、主目的は誤差 84.66% → 0.01%。

迷った案と採らなかった理由

ポイント解説

1. 「新しい」は 2 つある: 起きた時刻と届いた時刻。水位は届いた時刻、集計は起きた日
2. 足し算より置き換え: 影響日を丸ごと再計算すれば重複・順序・再実行に強い。L と週次窓で読む量に上限を置く
3. 欠陥は掛け算: 1 点直しても残る。JST だけで 3.07% でも潜在バグは残る
4. 日合計が合っても割当は違う: UTC は日合計 0.58% のずれで 16.86% を別日に付ける
5. dbt のテストは集計の正しさを見ない: 独立した再計算との照合が要る
6. 窓は分布から: L=3 は p99 を超え p99.9 は超えない。残りは週次

扱わなかった範囲（止め方）

• dbt・BigQuery は未実行。partitions の渡し方・枝刈り・require_partition_filter は記憶に基づく未確認
• 日次売上のみ。会員単位・キャンペーン単位、返品、税、スキーマ変更は扱っていない
• 遅延分布（92% 平均 1 分／6.5% 10 分〜12 時間／1.35% 12〜72 時間／0.15% 3〜10 日）と再配信 1.5% は仮定。自チームの loaded_at − event_ts を測って L・W を決める
• 1 日 2 万件一定の合成データなので、曜日変動やセール急増時の誤差は未測定
• Beam/Flink の allowed lateness や dbt microbatch など実物の仕様は、概念の対応づけまでで引用して確かめていない

発展: 会員別の累計値引き額を同じ方式で保つとき、窓外の遅延で累計がずれる。作り直す案と差分更新案を、誤差・費用・再実行の容易さで比べよ。

今日のまとめ

水位は届いた時刻、影響日を丸ごと再計算、重複は event_id で除去、日付は JST。窓外は照合で検知し週次で直す。1 点ずつ直しても誤差は残る（84.66% → 0.01%、合成データの実測）。この規模ならフルリフレッシュも有力。

自己評価（あとで記入）