生データを使える形に育てる — メダリオン設計と宣言的パイプライン

9.1メダリオン設計の実務判断 — Bronze で加工しない理由、overwrite の罠、Silver の dedup 過剰、Gold の事前集計

Bronze で「ちょっとだけ」整形すると、後から取り返しがつかない

取り込みの担当者が善意で「空文字は NULL に寄せておきました」「日付は yyyy-MM-dd に揃えておきました」とやってしまうことがあります。半年後にソース側が空文字と NULL を別の意味で使い分けていたと分かっても、もう元には戻せません。Bronze に元の姿が残っていないからです。メダリオン設計で Bronze を素通しにするのは美学の問題ではなく、後からやり直す余地を残すためです。公式ガイドも Bronze を検証を伴わない生データ取り込みの層と定義し、想定外のスキーマ変更でデータを失わないよう、大半のフィールドを string、VARIANT、binary のいずれかで保存することを勧めています。

ただし「何も足すな」という意味ではありません。取り込み時刻や取り込み元のファイル名といった来歴のメタデータは、公式にも Bronze へ任意で足してよい列として挙げられています。避けるべきなのは業務ロジックの側です。Bronze で WHERE status = 'active' のような行の絞り込みや、通貨換算のような列の計算を入れてしまうと、業務ルールが変わったときに過去分を作り直せません。絞り込みと計算は Silver 以降に集めます。

判断に迷ったら「この処理を入れると元データに戻れなくなるか」で切り分けます。戻れなくなる処理は Bronze に置きません。

overwrite は定期実行と組み合わせた瞬間に破壊的になる

次のコードは一見なんの問題もなく、単発のテストでは正しく動きます。

bronze = spark.read.format("delta").load("/mnt/bronze")
silver = bronze.filter("status = 'active'").groupBy("category").count()
silver.write.format("delta").mode("overwrite").save("/mnt/silver")

これを毎時のジョブに載せた途端に壊れます。mode("overwrite") は書き込み先を毎回まるごと置き換えるので、Silver に残るのは「そのとき Bronze にあった分だけ」の集計です。Bronze 側で古いデータを別の場所へ退避したり、ソースが直近ぶんしか持っていなかったりすると、Silver から過去が静かに消えます。BI 利用者から出る「先月の数字が今日になって変わった」はここから生まれます。

置き換え先は要件で決まります。追記だけでよいなら mode("append")、キーで突き合わせて更新も要るなら MERGE INTO、そもそも再計算を処理系に任せたいならマテリアライズドビューにします。overwrite が正当なのは、毎回ソース全体から決定的に同じ結果を作り直せると保証できるときだけです。マスタの洗い替えのような、ソースが常に全量を持つケースがこれに当たります。

この事故はテストでは再現しません。1 回だけ流せば結果は正しく、壊れるのは 2 回目以降だけです。定期実行に載せる前に、同じジョブを 2 回続けて流したら何が起きるかを必ず確認します。

Silver の重複排除は「効きすぎ」を先に疑う

Gold の売上が想定よりはるかに小さい、という相談で最初に見るのは Silver の dropDuplicates() です。三層のうち行数を意図的に減らす層は Silver しかないので、集計値が縮む原因はほぼここに集まります。よくあるのはキーが粗すぎる場合で、注文明細に対して dropDuplicates(["order_id"]) と書くと、1 注文に複数明細がある正当なデータまで 1 行に潰れます。["order_id", "line_no"] のように業務上の一意キーまで下ろす必要があります。削った件数は必ず記録します。

公式ガイドが Silver の作業として挙げているのは、重複排除、スキーマ強制、NULL や欠損値の扱い、順序が乱れたデータの解決、そして結合です。いずれも行数か値を変える操作なので、レビューではキーの粒度と結合の種類を必ず確認します。結合キーに NULL が混ざっている場合も要注意で、SQL の仕様どおり NULL = NULL は真になりません。内部結合で静かに行が消えるので、結合の前に WHERE key IS NOT NULL で明示的に落とし、落とした行は別テーブルに退避して見える化します。coalesce(key, 0) のような穴埋めは、無関係な行同士を結び付けてしまうため使いません。

Gold は読む側の形に寄せる

Gold は BI ダッシュボードやレポートが直接参照する層で、公式にはディメンショナルモデリングと集計の層と定義されています。ここで初めて列名をビジネス用語に直し、KPI を事前計算します。集計を Gold に固定しておくと、ダッシュボードごとに違う定義の「売上」が生まれる事故を防げます。逆に Bronze の生データを Gold へ直接ロードする設計は、型不整合やノイズを BI 利用者にそのまま見せることになるため採りません。

BI からの絞り込みが日付や地域に集中するなら、絞り込みに使う列でファイルの中身を並べ替えておく仕組み (Liquid Clustering) を使い、CLUSTER BY (order_date, region) のように指定します。クラスタリングキーは最大 4 列までで、パーティショニングや ZORDER とは併用できません。Delta テーブルでは Databricks Runtime 15.4 LTS 以上で一般提供です。ファイル配置の設計そのものは第 14 章で扱います。

メダリオンは Databricks が推奨する設計パターンであって、必須要件ではありません。公式ドキュメントも「推奨されるベストプラクティスであって要件ではない」と明記しています。三層に分ける狙いは、生データの保全、品質責任の集約、読み手向けの最適化という三つの関心事を混ぜないことにあります。

9.2更新をどう反映するか — MERGE INTO と Change Data Feed（readChangeFeed / startingVersion / table_changes()）

「毎回 Bronze を全件読み直している」から抜け出す

Bronze が数億行に育つと、Silver を作る夜間ジョブが 40 分から 2 時間へと伸びていきます。中身を見ると、変わったのは数万行なのに毎回全件を読み直しているだけ、というのがよくある姿です。ここから抜けるには二つの道具が要ります。差分を「当てる」側の MERGE INTO と、差分を「取り出す」側の Change Data Feed です。

MERGE INTO の三つの WHEN 句

MERGE INTO は Delta Lake テーブルでのみサポートされる文で、Parquet や CSV のテーブルには使えません。分岐は三種類あります。

MERGE INTO silver AS t
USING updates AS s
ON t.order_id = s.order_id
WHEN MATCHED AND s.op = 'D' THEN DELETE
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
WHEN NOT MATCHED BY SOURCE THEN UPDATE SET t.is_active = false;

WHEN MATCHED は両方にある行で UPDATE か DELETE、WHEN NOT MATCHED (別名 WHEN NOT MATCHED BY TARGET) はソースにだけある行で INSERT、WHEN NOT MATCHED BY SOURCE はターゲットにだけ残っている行で UPDATE か DELETE を書けます。三種類とも複数書けますが、条件を省略できるのは各種類の最後の 1 つだけです。それ以外の句に条件を付け忘れるとエラーになります。上の例で WHEN MATCHED THEN UPDATE SET * を先に置くと、後ろの削除分岐は永久に到達しません。1 行につき実行される分岐は 1 つだけで、上から順に最初に当たったものが採用されます。「一致した行に MATCHED と NOT MATCHED の両方が走る」ことはありません。

失敗のしかたも二つ覚えておきます。一つ目は、1 つのターゲット行に複数のソース行が一致した場合で、DELTA_MULTIPLE_SOURCE_ROW_MATCHING_TARGET_ROW_IN_MERGE でトランザクションごと失敗します。どちらの値で更新すべきか決められないためで、「最初の 1 行を採る」ような暗黙の挙動はありません。ソース側を事前に重複排除するか、キーごとに最新 1 行へ畳んでから渡します。二つ目はソースに新しい列が増えた場合で、既定では取り込まれません。Databricks Runtime 15.2 以降なら MERGE WITH SCHEMA EVOLUTION INTO ... と書いてターゲット側のスキーマを追随させられます。なお主キー制約は Databricks では情報目的の宣言にすぎず、MERGE INTO の前提条件ではありません。列名がソースとターゲットで違っていても UPDATE SET t.amount = s.amt のように個別に対応付けられます。

差分を取り出す — Change Data Feed

MERGE は「差分がすでに手元にある」ことを前提にします。その差分を Delta テーブル自身から取り出すのが Change Data Feed (CDF) です。既定では無効なので、明示的に有効化します。

ALTER TABLE bronze_orders
SET TBLPROPERTIES (delta.enableChangeDataFeed = true);

CREATE TABLE の TBLPROPERTIES に同じ指定を書いても構いません。ここで一番間違えやすいのは、有効化した時点より前の変更は取得できないことです。CDF は過去に遡って履歴を再構成する機能ではありません。

読み出しは二通りあります。バッチなら SQL 関数で範囲を指定します。

SELECT * FROM table_changes('bronze_orders', 5, 12);

ストリーミングやデータフレーム API なら、リーダー側のオプションで指定します。

(spark.readStream.format("delta")
   .option("readChangeFeed", "true")
   .option("startingVersion", 5)
   .table("bronze_orders"))

読み取りオプションは readChangeFeed、開始位置は startingVersion か startingTimestamp、終了位置は endingVersion か endingTimestamp です。タイムスタンプは 'yyyy-MM-dd[ HH:mm:ss[.SSS]]' の文字列で渡します。選択肢に並ぶ enableCDF、changeDataFeed、startVersion、fromVersion、delta.enableCDC はいずれも実在しません。

返ってくる行には元の列に加えて 3 列が付きます。_change_type、_commit_version、_commit_timestamp です。自動で付く一意な行 ID のような列はありません。_change_type が取る値は insert、update_preimage、update_postimage、delete の 4 つで、UPDATE は更新前と更新後の 2 行として現れます。この 2 行をそのまま MERGE に流すと更新前の値で上書きしてしまうので、update_preimage を落としてから渡します。

Time Travel との違いも整理します。VERSION AS OF や RESTORE TABLE が返すのは「その時点のテーブル全体のスナップショット」で、行が挿入されたのか更新されたのかは分かりません。DESCRIBE HISTORY が返すのは操作のメタデータだけで、行の中身は含みません。行レベルで「何がどう変わったか」を返すのは CDF だけです。監査とロールバックには Time Travel、増分 ETL には CDF、と役割で選びます。

二つをつないだ増分 ETL

Bronze の全件上書きは、次の形に置き換わります。前回処理したバージョンの次から読み、update_preimage を除き、キーごとに最新 1 行へ絞ってから MERGE INTO でぶつけます。この絞り込みが複数行一致エラーも防ぎます。

MERGE INTO silver_orders t
USING (SELECT * FROM table_changes('bronze_orders', 43)
       WHERE _change_type != 'update_preimage'
       QUALIFY row_number() OVER (PARTITION BY order_id
               ORDER BY _commit_version DESC) = 1) s
ON t.order_id = s.order_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

再実行しても同じ結果になるようにする

変更データは通常のデータファイルとは別に _change_data ディレクトリに書かれますが、書かれない操作もあります。追記だけの書き込みとパーティション丸ごとの削除は、既存のデータファイルから変更内容を復元できるため、専用ファイルを作りません。

運用で効いてくるのは保持期間です。変更データファイルもテーブルの保持ポリシーに従い、VACUUM で消えます。VACUUM の既定の保持しきい値は 7 日なので、それより古いバージョンから読もうとすると失敗します。長く止まっていたパイプラインを再開するときの典型的な落とし穴です。なお保持を既定より短くしようとすると安全チェックに弾かれ、spark.databricks.delta.retentionDurationCheck.enabled を false にしない限り実行できません。実行中のジョブがまだコミットしていないファイルを消す危険があるためです。

冪等性は自分で作ります。実行のたびに読む範囲を startingVersion と endingVersion で固定し、次回の開始位置を状態テーブルかチェックポイントに記録します。範囲が固定されていれば、やり直しても同じ差分が返り、下流の MERGE INTO がキーで吸収します。

もう一つ、列の改名、削除、型の変更といった非追加的なスキーマ変更をまたいで CDF を読むことはできず、クエリが失敗します。スキーマを変えるときは、変更の前後で読み取り範囲を切り、下流を一度追いつかせてから進めます。

9.3Lakeflow Spark Declarative Pipelines 実務 — @dlt.table / view、マテリアライズドビューとストリーミングテーブル、dlt.read と read_stream

「どこから流し直すか」を人間が考えている間は、まだ手続き的です

夜間の ETL が 3 本目のノートブックで落ちたとき、4 本目から流し直してよいのか、2 本目まで巻き戻すべきなのかを毎回考えることになります。この判断こそが手続き的な書き方のコストです。Lakeflow Spark Declarative Pipelines (旧 Delta Live Tables) は、テーブルを「どう作るか」ではなく「何であるか」で書かせることで、この判断を処理系に渡します。書き手はテーブルを定義する関数を並べるだけで、依存関係の解決、実行順序、更新、リトライはランタイムが引き受けます。

依存関係は明示的に書きません。ある関数が別のテーブルを読んでいれば、それが辺になって有向グラフが組み上がります。だから定義の記述順は結果に影響しませんし、途中で失敗しても「どこまで済んだか」はランタイムが把握しています。ストリーミングテーブルごとのチェックポイントもパイプラインが管理するので、writeStream を自分で書くときに必須だった checkpointLocation の指定は不要になります。

デコレータではなく戻り値でテーブル種別が決まる

現行の Python API は from pyspark import pipelines as dp でインポートします。従来の import dlt も動きますが、資料やサンプルは新しい名前に移りつつあります。

from pyspark import pipelines as dp

@dp.materialized_view
def customers():
    return spark.read.table("bronze_customers")

@dp.table
def events():
    return spark.readStream.table("bronze_events")

ここが試験で一番よく問われるところです。テーブルの種別はデコレータの引数ではなく、関数が返すデータフレームが静的かストリーミングかで決まります。公式ドキュメントも「マテリアライズドビューとストリーミングテーブルの基本構文の違いは spark.read か spark.readStream かだけ」と書いています。@dp.table(mode="live")、@dp.view(streaming=True)、type="streaming_table" のような引数は存在しません。

三つのデータセット種別を使い分ける

Kafka や Auto Loader からの取り込みを 1 分粒度で回したいならストリーミングテーブル、日次の集計マートならマテリアライズドビューが素直な選択です。三つ目の種別がビューで、オンデマンドで評価され永続化されません。カタログにも公開されないので、同じ式を二つのテーブルで使い回したい、品質チェック専用の中間段を挟みたい、といった用途に向きます。

SQL で書くときの対応

CREATE OR REFRESH MATERIALIZED VIEW gold_sales
AS SELECT region, sum(amount) AS amount
   FROM silver_events GROUP BY region;

CREATE OR REFRESH STREAMING TABLE silver_events (
  CONSTRAINT valid_id EXPECT (order_id IS NOT NULL) ON VIOLATION DROP ROW
)
AS SELECT * FROM STREAM(bronze_events) WHERE amount >= 100;

Python の spark.readStream に相当するのが SQL の STREAM(...) です。STREAM() を外すとバッチ読み込みになり、ストリーミングテーブルとして宣言していても増分にはなりません。

実行モードは Triggered と Continuous の 2 つ

宣言的に書いても、いつ動くかは自分で決めます。パイプラインの実行モードは二つです。Triggered は、更新を開始した時点で利用可能なデータを処理し終えたら停止します。10 分から 1 日おきの更新に向き、クラスタが動くのは処理中だけなので安価です。Continuous は止めるまで動き続け、ソースに届いたデータを到着順に処理します。10 秒から数分の鮮度が要るときに選びます。

重要なのは、マテリアライズドビューもストリーミングテーブルも、パイプラインの更新が走ったときにしか変わらないことです。Triggered モードでスケジュールが疎なら、ソースがどれだけ変わっても次回の更新まで結果は古いままです。「マテリアライズドビューに最新のソースが反映されない」という相談の多くはこれで、テーブル定義にもソースの読み方にも問題はありません。

自分で writeStream を書く場合の trigger も同じ考え方です。trigger(availableNow=True) は「今ある分をまとめて処理して停止する」ので、スケジュール起動のバッチ用途に向きます。新しいファイルが届くたびに反映したいのに更新されない、という症状は、この設定で 1 回動いて止まっていることが原因になります。継続的に処理したいなら既定のトリガーか processingTime を使います。

更新されないときに見る順番

「ストリーミングテーブルのつもりで書いたのに、新しいデータが反映されない」。この症状には原因が複数あるので、順番に見ます。

1. ソースの読み方が spark.read のままになっていないか。静的読み込みは一度きりなので、Triggered モードで何度起動しても増分になりません。spark.readStream、SQL なら STREAM(...) に直します。
2. パイプラインの実行モードとスケジュール。Triggered なら次の更新まで反映されません。自分で書いたストリームなら availableNow=True で 1 回動いて停止していないかを見ます。
3. 定義を変えたのに古い結果が残っている場合。すでに処理された行は自動では作り直されません。過去分もやり直すにはフルリフレッシュを実行します。フルリフレッシュはストリーミングテーブルのデータとフローのチェックポイントを消し、ソースの全レコードを処理し直します。

パーティション指定の有無、Unity Catalog の登録、SQL ウェアハウスの種別は、この症状の原因にはなりません。そこから疑い始めると遠回りになります。なおフルリフレッシュが安全なのは、ソースに再構築に必要なデータが残っている場合だけです。Kafka の保持期間が短い、オブジェクトストレージのライフサイクルでファイルが消えている、といった状況では対象テーブルから行が落ちます。Bronze を素通しで残す意味はここにもあります。

四つ目の可能性は、ソースが追記のみでなくなったときです。上流で行が更新・削除されるようになるとストリーミングテーブルの前提が崩れます。これは書き方の問題ではないので、次節の AUTO CDC で「変更として適用する」形に設計を切り替えます。

LIVE スキーマは黙って無視される

古い資料には LIVE.bronze や dlt.read("bronze") という書き方が出てきます。LIVE スキーマはレガシー公開モードで、同じパイプライン内の他のデータセットを参照するための仕組みでした。現在は新しいパイプラインでは LIVE スキーマの構文は黙って無視されます。エラーにならないので気付きにくい点に注意してください。新しく書くなら、パイプライン内の他のデータセットも通常のテーブルと同じく spark.read.table("customers") や spark.readStream.table("events") で参照します。

試験でもドキュメントでも新旧が混在するので、対応を読み替えられるようにしておきます。

9.4Expectations と apply_changes — expect 3種と ON VIOLATION、テーブル制約との使い分け、quarantine の扱い、CDC と SCD

NULL の主キーが Gold まで届いてから気付くことになります

ダッシュボードの合計が合わないと連絡が入り、追いかけると Bronze に order_id が NULL の行が数千件混ざっていて、Silver の結合ですべて落ちていた、という事故が起きます。取り込みの時点で気付けたはずのものです。Lakeflow Spark Declarative Pipelines の Expectations は、この「気付けたはず」を宣言としてパイプラインの中に埋め込む仕組みです。

expect / expect_or_drop / expect_or_fail の三択

Python はデコレータ、SQL は CONSTRAINT ... EXPECT 句で書きます。挙動は三つです。

ON VIOLATION を省略したときの既定は warn で、違反行はそのままターゲットに書き込まれます。「宣言しておけば勝手に弾いてくれる」と思い込むと、品質ルールを書いたつもりで何も止めていない状態になります。FAIL UPDATE は逆に厳しく、公式には「不正なレコードが更新の成功を妨げる」とされ、再処理の前に手動の介入が必要です。

ルールが増えたら辞書でまとめて渡せます。次のように書くと、二つの条件のどちらかに違反した行がターゲットから除外されます。

@dp.table
@dp.expect_all_or_drop({"valid_id": "order_id IS NOT NULL",
                        "valid_amount": "amount > 0"})
def silver_orders():
    return spark.readStream.table("bronze_orders")

@dp.expect_all、@dp.expect_all_or_drop、@dp.expect_all_or_fail が対応する 3 種です。選択肢に出てくる AUTO REPAIR、MASK COLUMN、ON VIOLATION QUARANTINE はいずれも実在しません。適用できる対象はストリーミングテーブル、マテリアライズドビュー、一時ビューに限られ、パイプラインの外にある通常の Delta テーブルには付けられません。違反件数はイベントログに記録され、パイプライン UI の Data quality タブから確認できます。メトリクスが出ないときは、述語の構文とテーブル定義への紐付けを最初に疑います。Expectations はテーブル定義の中に書いてはじめて評価されるので、別の関数に置くと何も起きません。Photon の有無やカタログの種類は関係ありません。

quarantine は標準のアクションではない

違反行を捨てずに別テーブルへ隔離したい、という要件は頻出しますが、ON VIOLATION にその挙動はありません。解き方は二つあります。一つは、同じソースから 2 本のテーブルを定義して述語を反転させる二系統パターンです。本流には expect_or_drop("valid", "order_id IS NOT NULL")、隔離側には expect_or_drop("invalid", "NOT (order_id IS NOT NULL)") を当てます。もう一つは warn の expect のまま全行を書き込んでおき、下流のビューで WHERE NOT (rule) と条件を反転して隔離テーブルを切り出す方式です。隔離は Expectations の機能ではなく、テーブル定義を分ける設計パターンだと理解しておきます。

Expectations とテーブル制約はレイヤが違う

Delta テーブル側にも品質の仕掛けがあります。ALTER TABLE orders ADD CONSTRAINT amt_pos CHECK (amount > 0); と書けば、違反する書き込みはトランザクションごと失敗します。NULL の禁止は ALTER TABLE orders ALTER COLUMN order_id SET NOT NULL; で、後から付けることもできます。ただし制約を追加する時点で既存行がすべて条件を満たすか検証されますので、汚れたテーブルには後付けできません。

CHECK と NOT NULL は強制される制約で、選択肢は通すか失敗させるかの二つだけです。Expectations は warn / drop / fail を選べる代わりに、パイプラインの外からの書き込みには効きません。実務では、業務上 NULL であってはいけない少数の列に NOT NULL と CHECK を絞って付け、細かいルールは Expectations 側に置きます。全列に一律で NOT NULL を付ける設計は、NULL が正当な列まで弾くため採りません。なお主キー、外部キー、UNIQUE は情報目的の制約で強制されないので、これらで品質を守ることはできません。

CDC は AUTO CDC に任せる

CDC ソースを取り込むとき、MERGE を手で書くと順序逆転と重複の処理を自分で作り込むことになります。宣言的パイプラインでは AUTO CDC (旧 APPLY CHANGES) がそれを引き受けます。ターゲットを先に宣言してからフローを作ります。

dp.create_streaming_table("silver_customers")

dp.create_auto_cdc_flow(
    target = "silver_customers",
    source = "bronze_cdc",
    keys = ["customer_id"],
    sequence_by = col("sequence_num"),
    apply_as_deletes = expr("operation = 'DELETE'"),
    except_column_list = ["operation", "sequence_num"],
    stored_as_scd_type = 2,
)

必須は target、source、keys、sequence_by の 4 つです。keys が行を識別し、sequence_by が同じキーに複数のイベントが来たときの順序を決めます。これがあるので、到着順が入れ替わっても最終状態が壊れません。ignore_null_updates の既定は False で、更新イベントの NULL は既存の値を NULL で上書きします。削除イベントを行の削除として扱うのが apply_as_deletes、テーブル全体の消去として扱うのが apply_as_truncates です。ターゲットに載せる列は column_list で挙げるか except_column_list で除きます。除いた 2 列は CDC の制御情報だからです。

SQL では、ターゲットのストリーミングテーブルを先に作り、フローとして宣言します。

CREATE OR REFRESH STREAMING TABLE silver_customers;

CREATE FLOW customers_cdc AS AUTO CDC INTO silver_customers
FROM STREAM(bronze_cdc)
KEYS (customer_id)
APPLY AS DELETE WHEN operation = 'DELETE'
SEQUENCE BY sequence_num
COLUMNS * EXCEPT (operation, sequence_num)
STORED AS SCD TYPE 2;

SCD Type 1 と Type 2

stored_as_scd_type の既定は "1" です。Type 2 で自動追加される 2 列のデータ型は sequence_by に指定した列と同じになります。履歴を追う列を絞りたいときは track_history_column_list か track_history_except_column_list を使います。

AUTO CDC のターゲットを同じパイプライン内から直接ストリーム読みすることはできません。更新も削除も起きるテーブルだからで、下流に流すときはターゲットの change feed を読みます。

品質は三段構えで守ります。テーブル制約で絶対に許さない値を止め、Expectations で許容範囲を宣言して件数を観測し、AUTO CDC で更新の適用そのものを宣言に寄せる。手で書いた MERGE と手で書いた WHERE の組み合わせは、壊れたときに何が起きたかを後から説明できません。