DatabricksのMERGE INTOとAPPLY CHANGES INTOの違い

どちらもSQLを用いてテーブルに対するUPSERT(UPDATE + INSERT)をするのですが、用途や仕様が異なるのでこちらにまとめます。

# MERGE INTO

https://docs.databricks.com/ja/delta/merge.html

**目的**

ターゲットのDeltaテーブルにソーステーブルからのデータをマージするための汎用SQL文です。UPSERT処理という観点では、後述の`APPLY CHANGES INTO`よりも汎用的です。

**用途**

マッチング条件に基づいたUPSERT、DELETE、UPDATEのように様々なオペレーションで使用することができます。

**複雑性**

順番がバラバラなデータを適切に適用するにはより複雑なロジックを必要とし、適切に対応しない場合には不適切な結果を生成することがあります。

**柔軟性**

マージの条件やアクションに対してよりきめ細かいコントロールが可能であり、重複排除は条件に基づく更新のような複雑な操作が可能です。

**構文**

```sql
MERGE INTO target_table_name [target_alias]
USING source_table_reference [source_alias]
ON merge_condition
WHEN MATCHED [ AND matched_condition ] THEN matched_action
WHEN NOT MATCHED [BY TARGET] [ AND not_matched_condition ] THEN not_matched_action
WHEN NOT MATCHED BY SOURCE [ AND not_matched_by_source_condition ] THEN not_matched_by_source_action;
```

**SQLリファレンス**

https://learn.microsoft.com/ja-jp/azure/databricks/sql/language-manual/delta-merge-into


# APPLY CHANGES INTO

https://docs.databricks.com/ja/delta-live-tables/cdc.html

**目的**

Delta Live Tablesパイプラインにおけるチェンジデータキャプチャ(CDC)のために特別に設計されたSQL文です。Delta Live Tables外では使用できません。

**用途**

[チェンジデータフィード(CDF)](https://docs.databricks.com/ja/delta/delta-change-data-feed.html)やスナップショットからの変更点を処理するために活用されます。

**順番がバラバラのデータの取り扱い**

CDCシナリオにおいてよくある問題である順番がバラバラのレコードを取り扱えるのように最適化されています。

**シンプルさ**

変更点を適用するために、より分かりやすく宣言型であるアプローチを提供しており、CDCロジックの実装を容易にします。

**SCDタイプのサポート**

[Slowly Changing Dimensions (SCD)](https://docs.databricks.com/ja/delta-live-tables/cdc.html#example-scd-type-1-and-scd-type-2-processing-with-cdf-source-data)のタイプ1とタイプ2をサポートしており、直接的なアップデートと変更点の履歴追跡を可能にします。


**構文**

```sql
CREATE OR REFRESH STREAMING TABLE target;
APPLY CHANGES INTO live.target
FROM stream(cdc_data.users)
KEYS (userId)
APPLY AS DELETE WHEN operation = "DELETE"
SEQUENCE BY sequenceNum
COLUMNS * EXCEPT (operation, sequenceNum)
STORED AS SCD TYPE 1;
```

2つの違いを表にまとめます。

|  | MERGE INTO | CHANGE APPLY INTO |
|:--|:--|:--|
| 用途  | 汎用的なマージ処理  | チェンジデータキャプチャ(CDC)  |
| どこで利用できるか  | Delta Live Tables以外  | Delta Live Tables  |
| 長所  | <ul><li>ソーステーブルとターゲットテーブルでマッチングする列名が異なっていても対応可能。</ul>  | <ul><li>順序がバラバラなデータであっても`SEQUENCE BY`句で順序を守って更新することができる。 <li>ソーステーブルの増分レコードに基づいて更新することができる。</ul> |
| 短所  |  <ul><li>複数のレコードがマッチング条件に該当するとエラーになる。<li>順序がバラバラのデータで適切にマージを行う際には追加のロジック実装外必要。</ul> | <ul><li>マッチングを行う列名は同名でなくてはならない。<li>`APPLY CHANGES INTO`で直接ソーステーブルとは異なる列名をターゲットテーブルに指定することはできない。前段でビューを用いるなどして変更する必要がある。</ul>  |

# 例

上述の違いを体験するために、ここではまずは`MERGE INTO`でUPSERTを実装し、同様の処理を`APLLY CHANGE INTO`で実装します。

## `MERGE INTO`の実装

以下のようなソーステーブルを作成します。

![Screenshot 2024-12-10 at 16.44.21.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/50069169-f2b5-90e5-fe20-b30780aff794.png)

`STORE_ID`と`PRODUCT_ID`に基づいてターゲットテーブルを更新するシナリオを考えます。ただし、ターゲットテーブルでは異なる列名を用いており、`TENPO_ID`と`SEIHIN_ID`となっています。

基本的なMERGE INTOの構文の用法は以下のようになります。

```sql
CREATE TABLE takaakiyayoi_catalog.cdc.target(TENPO_ID INT, SEIHIN_ID STRING, SEIHIN_MEI STRING)
```


```sql
MERGE INTO takaakiyayoi_catalog.cdc.target AS target USING (
  SELECT
    *
  FROM
    takaakiyayoi_catalog.cdc.source
) AS source ON target.TENPO_ID = source.STORE_ID
AND target.SEIHIN_ID = source.PRODUCT_ID
WHEN matched then
update
set
  target.TENPO_ID = source.STORE_ID,
  target.SEIHIN_ID = source.PRODUCT_ID,
  target.SEIHIN_MEI = source.PRODUCT_NAME
  WHEN NOT matched then
insert
  (TENPO_ID, SEIHIN_ID, SEIHIN_MEI)
VALUES
  (
    source.STORE_ID,
    source.PRODUCT_ID,
    source.PRODUCT_NAME
  )
```

ソーステーブルとターゲットテーブルでマッチングする列が異なっていても`ON`句で吸収することができます。しかし、この例ではこのSQLを実行するとすべての行がINSERTされます。

![Screenshot 2024-12-10 at 16.55.49.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/0a642813-b4e4-2116-f72e-f55f61cf87c4.png)

このため、複数一致が発生しないように以下のように`WHERE UPDATE_DATE = "2024-10-01"`を追加します。

```sql
MERGE INTO takaakiyayoi_catalog.cdc.target AS target USING (
  SELECT
    *
  FROM
    takaakiyayoi_catalog.cdc.source
  WHERE
    UPDATE_DATE = "2024-10-01"
) AS source ON target.TENPO_ID = source.STORE_ID
AND target.SEIHIN_ID = source.PRODUCT_ID
WHEN matched then
update
set
  target.TENPO_ID = source.STORE_ID,
  target.SEIHIN_ID = source.PRODUCT_ID,
  target.SEIHIN_MEI = source.PRODUCT_NAME
  WHEN NOT matched then
insert
  (TENPO_ID, SEIHIN_ID, SEIHIN_MEI)
VALUES
  (
    source.STORE_ID,
    source.PRODUCT_ID,
    source.PRODUCT_NAME
  )
```

これで、`2024-10-01`のみのデータに基づいてUPSERT(実際にはINSERT)が行われました。

日付を変えて再度実行します。

```sql
MERGE INTO takaakiyayoi_catalog.cdc.target AS target USING (
  SELECT
    *
  FROM
    takaakiyayoi_catalog.cdc.source
  WHERE
    UPDATE_DATE = "2024-10-02"
) AS source ON target.TENPO_ID = source.STORE_ID
AND target.SEIHIN_ID = source.PRODUCT_ID
WHEN matched then
update
set
  target.TENPO_ID = source.STORE_ID,
  target.SEIHIN_ID = source.PRODUCT_ID,
  target.SEIHIN_MEI = source.PRODUCT_NAME
  WHEN NOT matched then
insert
  (TENPO_ID, SEIHIN_ID, SEIHIN_MEI)
VALUES
  (
    source.STORE_ID,
    source.PRODUCT_ID,
    source.PRODUCT_NAME
  )
```

![Screenshot 2024-12-10 at 16.57.35.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/b119cce8-ea6d-1ed8-926d-3dec7dd39003.png)

このように、単にマージ(UPSERT)処理を行いたいというケースであれば、柔軟な`MERGE INTO`を使った方が優れていると言えます。

では、`APPLY CHANGES INTO`の例を見てみましょう。

## `APPLY CHANGES INTO`の実装

こちらは、Delta Live Tablesで実装するので、ノートブックを作成し、以下を記述します。

```sql
CREATE OR REFRESH STREAMING TABLE dlt_target;
```

```sql
APPLY CHANGES INTO
  live.dlt_target
FROM
  stream(takaakiyayoi_catalog.cdc.source)
KEYS
  (STORE_ID, PRODUCT_ID) -- sourceのキー
SEQUENCE BY
  UPDATE_DATE -- 更新順序を表す列
STORED AS
  SCD TYPE 1 -- SCDタイプ
```

このノートブックを指定してパイプラインを作成し、実行します。

![Screenshot 2024-12-10 at 17.03.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/a3459945-4fda-ccab-9108-a6b31e5330f1.png)

これで、`UPDATE_DATE`に基づいてUPSERTが行われます。

![Screenshot 2024-12-10 at 17.04.10.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/9d65394a-afa5-3c44-3f58-1bafc8ea737d.png)

しかし、この状態ではソーステーブルとターゲットテーブルの列名が同じままとなっています。`APPLY CHANGES INTO`では、列名が異なるテーブル同士のマッチングができないため、`APPLY CHANGES INTO`の前段に[一時ビュー](https://docs.databricks.com/ja/delta-live-tables/sql-ref.html#create-a-delta-live-tables-view)を挟んで列名を変更します。

```sql
CREATE OR REFRESH STREAMING TABLE dlt_target;
```

```sql
CREATE TEMPORARY STREAMING LIVE VIEW dlt_intermediate AS
SELECT
  STORE_ID AS TENPO_ID,
  PRODUCT_ID AS SEIHIN_ID,
  PRODUCT_NAME,
  UPDATE_DATE
FROM
  stream(takaakiyayoi_catalog.cdc.source)
```

```sql
APPLY CHANGES INTO
  live.dlt_target
FROM
  stream(live.dlt_intermediate)
KEYS
  (TENPO_ID, SEIHIN_ID) -- sourceのキー
SEQUENCE BY
  UPDATE_DATE -- 更新順序を表す列
STORED AS
  SCD TYPE 1 -- SCDタイプ
```

![Screenshot 2024-12-10 at 17.15.37.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/ab8ec7a2-c03f-8846-adf6-17f1c762558c.png)

期待した通りの挙動になりました。

![Screenshot 2024-12-10 at 17.15.55.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/cdeb986a-be2e-dfb5-e943-d0fe68445deb.png)

さらに以下のSQLを実行して、ソーステーブルにINSERTします。

```sql
INSERT INTO takaakiyayoi_catalog.cdc.source VALUES(1, "AA", "2024-10-3", "SANDWICH")
```

Delta Live Tablesのパイプラインを再度実行します。すると、1行のみが処理されます。

![Screenshot 2024-12-10 at 17.22.16.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/5ee0861c-f828-a6ec-e123-f43aa9a48436.png)

結果を確認すると、INSERTしたレコードに基づいてUPSERTが行われていることを確認できます。これが、ソーステーブルの変更データ(CHANGE DATA)のみを捕捉(CAPTURE)して更新を行う、**CHANGE DATA CAPTURE(CDC)** という訳です。データ量が多くなってくると、テーブル洗い替えのコストも無視できなくなってきますので、CDCの価値が出て来ると思います。

![Screenshot 2024-12-10 at 17.22.38.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4fdcdac9-3b9f-c6cd-6c36-5940b97bf24c.png)


# まとめ

`MERGE INTO`と`APPLY CHANGES INTO`の主な違いをまとめます。

1. **CDCへの最適化:** `APPLY CHANGES INTO`はCDCに最適化されています。ソーステーブルから絶え間なく流れ込むチェンジデータフィード(CDF)をインクリメンタルに処理し、ターゲットテーブルに適切に反映させる必要があるシナリオでは`APPLY CHANGES INTO`が適しています。`MERGE INTO`でこのような処理に対応しようとすると、追加のロジック実装が必要となります。
1. **シンプルさ vs. 柔軟性:** `APPLY CHANGES INTO`はCDCに特化している分、仕様はシンプルです。一方、`MERGE INTO`は複雑なデータ操作のための柔軟性とコントロールを提供しています。


### はじめてのDatabricks

[はじめてのDatabricks](https://qiita.com/taka_yayoi/items/8dc72d083edb879a5e5d)

### Databricks無料トライアル

[Databricks無料トライアル](https://databricks.com/jp/try-databricks)