Delta MERGE完全ガイド｜UPSERT・重複排除・SCD設計パターン

MERGE INTOは、Delta Lakeで最もよく使われるデータ操作のひとつです。 1つのステートメントで「マッチした行は更新、マッチしない行は挿入、不要な行は削除」ができるため、 UPSERT（Update or Insert）パターン、重複排除、SCD（Slowly Changing Dimension）の実装に広く使われます。 Databricks認定試験では全てのData Engineer系試験で頻出です。

基本構文

MERGE INTOの構文は、ターゲットテーブル・ソースデータ・マッチ条件・マッチ時の処理・非マッチ時の処理で構成されます。

MERGE INTO silver.customers AS t
USING bronze.raw_customers AS s
ON t.customer_id = s.customer_id

-- マッチした行: 更新
WHEN MATCHED AND s.updated_at > t.updated_at THEN
  UPDATE SET
    t.name = s.name,
    t.email = s.email,
    t.updated_at = s.updated_at

-- マッチしない行: 挿入
WHEN NOT MATCHED THEN
  INSERT (customer_id, name, email, created_at, updated_at)
  VALUES (s.customer_id, s.name, s.email, s.created_at, s.updated_at)

-- ソースにない行: 削除（オプション）
WHEN NOT MATCHED BY SOURCE THEN
  DELETE;

WHEN NOT MATCHED BY SOURCE はDelta Lake 2.4以降で追加された構文で、 ターゲットにはあるがソースにはない行（つまり削除されたデータ）を処理できます。

パターン1: UPSERT（Update or Insert）

最も基本的なパターンです。キーが一致すれば更新、一致しなければ挿入します。 Silver層へのデータ取り込みで最もよく使います。

-- シンプルなUPSERT
MERGE INTO silver.products t
USING staging.new_products s
ON t.product_id = s.product_id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

SET * と INSERT * は「全カラム」を意味します。 カラム構造がソースとターゲットで同じ場合に使えるショートカットです。

パターン2: 重複排除（Deduplication）

Bronze層には同じレコードが複数回到着することがあります（リトライ、重複送信など）。 MERGE + ROW_NUMBERで最新の1件だけを取り込む重複排除パターンが定番です。

MERGE INTO silver.events t
USING (
  SELECT *, ROW_NUMBER() OVER (
    PARTITION BY event_id
    ORDER BY event_timestamp DESC
  ) AS rn
  FROM bronze.raw_events
  WHERE _ingested_at > current_timestamp() - INTERVAL 1 HOUR
  QUALIFY rn = 1
) s
ON t.event_id = s.event_id
WHEN MATCHED AND s.event_timestamp > t.event_timestamp
  THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

ソース側で事前にROW_NUMBERで重複を排除してからMERGEするのがポイントです。 MERGEのソースに重複があると、同一キーに対して複数のマッチが発生してエラーになります。

パターン3: SCD Type 1（上書き更新）

SCD Type 1は、変更があった場合に既存レコードを単純に上書きするパターンです。 履歴は保持しません。最新の状態だけが重要な場合に使います。

-- SCD Type 1: 変更があれば上書き
MERGE INTO dim.customers t
USING staging.customer_updates s
ON t.customer_id = s.customer_id
WHEN MATCHED AND (
  t.name != s.name OR
  t.email != s.email OR
  t.address != s.address
) THEN UPDATE SET
  t.name = s.name,
  t.email = s.email,
  t.address = s.address,
  t.updated_at = current_timestamp()
WHEN NOT MATCHED THEN INSERT *;

パターン4: SCD Type 2（履歴保持）

SCD Type 2は、変更前のレコードを「期限切れ」にし、変更後のレコードを新しい行として挿入するパターンです。 変更履歴を完全に保持します。is_current フラグと有効期間（effective_from / effective_to）で管理します。

-- SCD Type 2: ステップ1 - 既存レコードの期限切れ処理
MERGE INTO dim.customers t
USING staging.customer_updates s
ON t.customer_id = s.customer_id AND t.is_current = true
WHEN MATCHED AND (t.name != s.name OR t.email != s.email) THEN
  UPDATE SET
    t.is_current = false,
    t.effective_to = current_date();

-- SCD Type 2: ステップ2 - 新レコードの挿入
INSERT INTO dim.customers
SELECT
  customer_id,
  name,
  email,
  true AS is_current,
  current_date() AS effective_from,
  '9999-12-31' AS effective_to
FROM staging.customer_updates s
WHERE NOT EXISTS (
  SELECT 1 FROM dim.customers t
  WHERE t.customer_id = s.customer_id
    AND t.is_current = true
    AND t.name = s.name
    AND t.email = s.email
);

SCD Type 2はMERGEだけでは完結しにくいため、UPDATE + INSERT の2ステップで実装するのが一般的です。 Delta Live Tables（DLT）には APPLY CHANGES INTO というSCD Type 2専用の構文があり、 これを使うとかなり簡潔に書けます。試験ではこの構文も出題範囲です。

パターン5: foreachBatch + MERGE（ストリーミング）

Structured Streamingで増分データをSilver層にMERGEするには、foreachBatch を使います。 各マイクロバッチの結果をDataFrameとして受け取り、MERGEを実行します。

def upsert_to_silver(batch_df, batch_id):
    # ソース側で重複排除
    deduped = batch_df.dropDuplicates(["customer_id"])
    deduped.createOrReplaceTempView("updates")

    spark.sql("""
      MERGE INTO silver.customers t
      USING updates s
      ON t.customer_id = s.customer_id
      WHEN MATCHED THEN UPDATE SET *
      WHEN NOT MATCHED THEN INSERT *
    """)

(spark.readStream
  .format("delta")
  .table("bronze.raw_customers")
  .writeStream
  .foreachBatch(upsert_to_silver)
  .option("checkpointLocation", "/checkpoint/silver_customers")
  .trigger(availableNow=True)
  .start()
)

パフォーマンスの考慮点

試験で問われるポイント

• MERGE INTO の基本構文（ON / WHEN MATCHED / WHEN NOT MATCHED）
• SET * と INSERT * の意味
• ソースに重複がある場合のエラーと対策（ROW_NUMBER）
• SCD Type 1 と Type 2 の違い
• foreachBatch + MERGE のパターン
• WHEN NOT MATCHED BY SOURCE の意味
• MERGEとINSERT OVERWRITEの使い分け

問題で確認

よくある質問