変換（ELT・dbt・冪等性）

昔は、DWH に入れる前に専用のサーバーで変換する ETL（Extract → Transform → Load）が主流でした。クラウド DWH が安く速くなった今は、まず生のまま DWH に入れて、DWH の中で SQL で変換する ELT が主流です。生のデータが残るのでやり直しがきき、SQL が書ける人なら誰でも変換に参加できます。

dbt で変換をコードとして管理する

dbt は ELT の「T」を担う道具です。1つの SELECT 文を1つのモデル（テーブルやビュー）として書き、ref() で他のモデルを参照すると、依存関係の順に実行してくれます。SQL を Git で管理し、レビュー・テスト・ドキュメント作りができるようになります。

-- models/staging/stg_shop__orders.sql
SELECT
  id                          AS order_id,
  user_id                     AS customer_id,
  CAST(total AS NUMERIC)      AS amount,
  created_at                  AS ordered_at
FROM {{ source('shop', 'orders') }}
WHERE _deleted IS NOT TRUE

source() で元のテーブルを、ref() で他のモデルを参照します。テーブルの名前を直接書かないことで、依存関係が分かり、開発用と本番用でスキーマを切り替えられます。

dbt のモデルは、役割ごとに層に分けます。層を分けると、どこに何を書くかが決まり、変更の影響が読みやすくなります。

レイクハウスでは同じ考え方をメダリオン（bronze＝生データ、silver＝整えたデータ、gold＝利用者向け）と呼ぶことが多く、名前は違っても「層ごとに役割を分ける」点は同じです。

よくある失敗：staging に業務ロジック（「売上はキャンセルを除く」）を入れて、元の値が見られなくなる。mart から mart を何段も参照して、1つの変更の影響が読めなくなる。テストを書かずにモデルだけ増やす。最低でも、各モデルの主キーに unique と not_null を入れます。

dbt のテストは、モデルの YAML に書きます。よく使う4つは unique・not_null・accepted_values（決まった値だけか）・relationships（参照先に存在するか）です。

models:
  - name: fct_orders
    columns:
      - name: order_id
        tests: [unique, not_null]
      - name: status
        tests:
          - accepted_values:
              values: ['paid', 'shipped', 'cancelled']
      - name: customer_id
        tests:
          - relationships:
              to: ref('dim_customers')
              field: customer_id

dbt build はモデルとテストを依存の順に実行し、テストが失敗するとその下流のモデルは作りません。壊れたデータがダッシュボードまで届くのを止められます。

テストは「最低でも各モデルの主キーに unique と not_null」から始めます。テストの考え方や置き場所はデータのテストで詳しく扱います。

冪等（べきとう）とは、同じ処理を何回やっても結果が同じになることです。パイプラインは必ずどこかで失敗し、やり直されます。「途中まで INSERT してから落ちた処理をもう一度流したら、半分が2倍になった」という事故を防ぐには、最初から冪等に書きます。INSERT INTO … SELECT を足していくだけの処理は冪等ではありません。

MERGE INTO fct_orders AS t
USING stg_orders_delta AS s       -- 今回の差分（キーごとに1行に重複を除いておく）
  ON t.order_id = s.order_id
WHEN MATCHED THEN
  UPDATE SET status = s.status, amount = s.amount, updated_at = s.updated_at
WHEN NOT MATCHED THEN
  INSERT (order_id, status, amount, updated_at)
  VALUES (s.order_id, s.status, s.amount, s.updated_at);

MERGE の元（USING）に同じキーが2行あると、エラーになるか結果が不定になります。元を必ずROW_NUMBER で1行にしてから使います。

増分処理は、毎回全件を作り直す代わりに、新しく来た分・変わった分だけを処理する方法です。データが大きくなるほど、時間と料金の差が開きます。dbt では materialized='incremental' と unique_key を指定し、is_incremental() の中に「前回より新しい分」の条件を書きます。

{{ config(materialized='incremental', unique_key='order_id') }}

SELECT order_id, status, amount, updated_at
FROM {{ ref('stg_orders') }}
{% if is_incremental() %}
  -- 前回の最大より少し前から取り、重なりは unique_key の MERGE で吸収する
  WHERE updated_at >= (SELECT MAX(updated_at) FROM {{ this }}) - INTERVAL '1 day'
{% endif %}

遅れて届くデータへの備えは、イベントを扱う集計では欠かせません。スマホのアプリは電波がない間イベントをためて、あとでまとめて送ります。「昨日のパーティションだけ処理する」だと、今日届いた昨日の分が漏れます。対策は、毎回直近の数日分（例：3日）を作り直すことです（ルックバック）。「前回の最大の時刻より後」で絞るときも、その時刻から少し重ねて取り、重なった分は MERGE で吸収します。イベントが起きた時刻と届いた時刻の両方を持っておくと、どのくらい遅れて届くかを測って、さかのぼる日数を決められます。

変換の SQL もアプリのコードと同じように、開発環境 → レビュー → CI → 本番の流れで出します。本番のテーブルを手で直接直すと、何が変わったか誰にも分からなくなります。

DWH の中の変換は SQL を基本にし、SQL では難しいことだけを Python などで書くのが、保守しやすい分け方です。

Python で DWH から全件を読み出してメモリで処理すると、データが増えたときに必ず詰まります。大きな処理は SQL か Spark に寄せ、Python は「指示を出す」「SQL ではできない部分だけ」に絞ります。dbt や DWH には Python のモデルを DWH の中で動かす仕組みもあります。