Delta Lakeにダイビング：DMLの内部処理(Update、Delete、Merge)

[Delta Lake Tutorial: How to Easily Delete, Update, and Merge Using DML \- The Databricks Blog](https://databricks.com/blog/2020/09/29/diving-into-delta-lake-dml-internals-update-delete-merge.html)の翻訳です。

これまでの記事[トランザクションログを読み解く](https://qiita.com/taka_yayoi/items/b7f628c219463e055592)、[スキーマの強制、進化](https://qiita.com/taka_yayoi/items/0336b4bbf9b0daaf92d1)でDelta Lakeのトランザクションログがどのように動作するのか、スキーマ強制と進化の内部処理を説明しました。Delta Lakeは`DELETE`、`UPDATE`、`MERGE`を含むDML(Data manipulation language)コマンドをサポートしています。これらのコマンドは**チェンジデータキャプチャ(CDC)や監査、ガバナンス**、さらには、**GDPR/CCPAワークフロー**などをシンプルなものにします。この記事では、それぞれのDMLコマンドの使い方、コマンドを実行した際に背後でDelta Lakeが何をしているのかを説明し、それぞれに対するパフォーマンスチューニングのティップスを紹介します。特に以下の点をご説明します：

- Delta LakeのACIDトランザクションログ入門
- DELETE、UPDATE、MERGEを実行する際の基本を理解する
- これらのタスクを実行する際のアクションを理解する
- Delta Lakeにおけるパーティションプルーニングを理解する
- Delta Lakeでストリームがどのように動作するのか

この記事に関する情報は、[Diving into Delta Lake Part 3: How do DELETE, UPDATE, and MERGE work](https://www.youtube.com/watch?v=7ewmcdrylsA)の動画でも参照できます。
<iframe width="560" height="315" src="https://www.youtube.com/embed/7ewmcdrylsA" title="YouTube video player" frameborder="0" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture" allowfullscreen></iframe>

# Delta Lake: 基本的なメカニズム

まず初めに、Delta Lakeがファイルレベルでどのように構成されているのかをクイックに見てみましょう。新たなテーブルを作成した際、Deltaはデータを一連のParquetファイルとして保存し、Delta Lakeトランザクションログを`_delta_log`フォルダに作成します。ACIDトランザクションログは、テーブル作成以来の全ての変更(トランザクション)のマスターレコードとして動作します。テーブルに変更(例えば、新規データの追加、更新、マージ)を加えると、Delta Lakeはそれぞれの新たなトランザクションを、`_delta_log`フォルダ内に`00...00000.json`で始まり、カウントアップしていくJSONファイルとして保存します。10トランザクションごとに、Deltaは"チェックポイント"Parquetファイルを同じフォルダに作成します。これにより、読み取り処理がテーブルの状態を迅速に再構築できます。
![](https://databricks.com/wp-content/uploads/2020/09/blog-diving-delta-2.png)

最終的には、Delta　Lakeテーブルにクエリーを行った際、読み取りを行うリーダーはトランザクションログを参照し、迅速にどのデータファイルが最新バージョンのテーブルを構成しているのかを特定します。クラウドオブジェクトストレージからファイルの一覧を取得するのではなく、必要なファイルの正確なパスを指定することで、劇的なクエリー性能の改善を実現します。この記事で議論するように、DMLオペレーションによってDelta Lakeはその場でテーブルを変更するのではなく、新たなバージョンのファイルを作成します。そして、それを追跡するためにトランザクションログを使用します。詳細に関しては、以前の記事[Delta Lakeにダイビング #1：トランザクションログを読み解く](https://qiita.com/taka_yayoi/items/b7f628c219463e055592)を参照ください。

これでファイルシステムレベルでDelta Lakeがどのように動作するのかに関して基本的な理解ができたので、次にDelta LakeでどのようにDMLコマンドを使うのか、内部で何が起きているのかに関してダイブしましょう。以下の例では、Delta Lake 0.7.0とApache Spark 3.0におけるSQLシンタックスを用います。詳細に関しては、[Enabling Spark SQL DDL and DML in Delta Lake on Apache Spark 3\.0](https://databricks.com/blog/2020/08/27/enabling-spark-sql-ddl-and-dml-in-delta-lake-on-apache-spark-3-0.html)を参照ください。

# Delta Lake DML: UPDATE

**述語(predicate)**とも呼ばれるフィルタリング条件にマッチする行を選択的に更新するために`UPDATE`を使用できます。以下のコードは`UPDATE`文における述語の使い方を説明しています。

```sql
-- Update events
UPDATE events SET eventType = 'click' WHERE eventType = 'clck'    
```

## UPDATE: 内部処理

Delta　Lakeはテーブルに対する`UPDATE`を2つのステップで実行します：

1. 述語に合致するデータを含むファイルを検索し選択します。Delta Lakeはこのプロセスを高速化できる場合には[データスキッピング](https://docs.databricks.com/delta/optimizations/file-mgmt.html#data-skipping)を活用します。
1. マッチしたファイルをメモリに読み込み、適切な行をアップデートし、結果を新たなデータファイルに書き出します。
![](https://databricks.com/wp-content/uploads/2020/09/blog-diving-delta-3.gif)

Delta Lakeが`UPDATE`の処理に成功すると、以降は古いデータファイルの代わりに新たなデータファイルが使用されることを示すコミットをトランザクションログに追加します。しかし、古いファイルは削除されません。代わりに、"墓石を建てます"。最新バージョンではなく、古いバージョンのテーブルに適用されたデータファイルとして記録されます。Delta Lakeはこれをデータのバージョン管理やタイムトラベルに活用します。

## UPDATE + Delta Lakeのタイムトラベル = 容易なデバッグ

古いデータファイルを保持することで、テーブルの過去の任意のバージョンを参照できるDelta Lakeの「タイムトラベル」を利用できるので、デバッグが容易になります。誤ってテーブルをアップデートしてしまった際に、何が起きたのかを明らかにしたい場合には、二つのバージョンのテーブルを容易に比較することができます。

```sql
SELECT * FROM events VERSION AS OF 12
```

## UPDATE: パフォーマンスチューニングのティップス

Delta Lakeにおける`UPDATE`コマンドのパフォーマンスを改善する主たる方法は、検索空間を狭めるために、述語を追加するということです。検索が限定的であるほど、Delta Lakeがスキャンし、変更するファイルの数が少なくて済みます。

DatabricksのマネージドバージョンであるDelta Lakeには、改善された[データスキッピング](https://docs.databricks.com/delta/optimizations/file-mgmt.html#data-skipping)、ブルームフィルターの利用、マルチカラムソートの改善版とも言える[Z-Order Optimize](https://docs.databricks.com/delta/optimizations/file-mgmt.html#z-ordering-multi-dimensional-clustering)(多次元クラスタリング)など、さらなるパフォーマンス改善の機能が含まれています。Z-orderingは、効率を最大化するために、類似のカラム値が戦略的に隣り合わせに配置されるように、それぞれのデータファイルのレイアウトを再構成します。詳細は[こちら](https://docs.databricks.com/delta/optimizations/file-mgmt.html#z-ordering-multi-dimensional-clustering)を参照ください。

> **参考記事**
[Databricks Deltaを使って秒でペタバイトデータを処理する \- Qiita](https://qiita.com/taka_yayoi/items/4bd090229b5b13a533f5)

# Delta Lake DML: DELETE

述語(フィルタリング条件)に合致する行を選択的に削除する際に`DELETE`コマンドを使用できます。

```sql
DELETE FROM events WHERE date < '2017-01-01'
```

間違った`DELETE`操作を取り消したい場合には、以下のPythonコードにあるように、タイムトラベルを使用してテーブルを元の状態に戻すことができます。

```python
# Read correct version of table into memory
dt = spark.read.format("delta") \
               .option("versionAsOf", 4) \
               .load(deltaPath)
    
# Overwrite current table with DataFrame in memory
dt.write.format("delta") \
        .mode("overwrite") \
        .save(deltaPath)
```

## DELETE: 内部処理

`DELETE`は`UPDATE`と同様に動作します。Delta Lakeは2回データをスキャンします：最初のスキャンは、フィルタリング条件に合致する行を含むデータファイルを特定するためのものです。2番目のスキャンでは、マッチするデータファイルをメモリに読み込みます。その時点で、新たなクリーンなデータをディスクに書き込む前にDelta Lakeが該当の行を削除します。

Delta Lakeが`DELETE`処理を成功した後に、古いデータファイルは削除されません。ディスクに留まり続けますが、Delta Lakeトランザクションログには「墓石(アクティブなテーブルとして見做されません)」として記録されます。以前のバージョンのテーブルにタイムトラベルで戻せるようにするために、古いファイルは即時削除されないことに注意してください。特定期間を経過した古いファイルを削除する際には、`VACUUM`コマンドを使用します。

## DELETE + VACUUM: 古いデータファイルのクリーンアップ

`VACUUM`コマンドを実行すると、以下のデータファイル全てを恒久的に削除します。

1. アクティブテーブルではないデータファイル
1. 保持期間(デフォルトは7日)より古いデータファイル

Delta Lakeは自動で古いファイルの`VACUUM`を行いません。以下に示すように、ご自身でコマンドを実行する必要があります。デフォルトの7日の保持期間を変更したい場合には、パラメータとして指定します。

```python
from delta.tables import *

# vacuum files not required by versions older than the default
# retention period, which is 168 hours (7 days) by default
dt.vacuum()
deltaTable.vacuum(48) # vacuum files older than 48 hours
```

> **注意**
0時間の保持期間を指定したVACUUMの実行は、最新バージョンのテーブルで使用されていない全てのファイルを削除します。データ損失が起こる可能性があるので、このコマンドを実行する際には、テーブルに書き込み中の処理がないことを確認してください。

`VACUUM`コマンドの詳細、Scala、SQLのサンプルに関しては、[VACUUMコマンドのドキュメント](https://docs.delta.io/latest/delta-utility.html#-delta-vacuum)を参照ください。

## DELETE: パフォーマンスチューニングのティップス

`UPDATE`コマンドと同様に、Delta Lakeにおける`DELETE`オペレーションの性能を改善する主たる方法は、検索空間を狭めるために述語を追加するというものです。DatabricksのマネージドバージョンであるDelta Lakeには、改善された[データスキッピング](https://docs.databricks.com/delta/optimizations/file-mgmt.html#data-skipping)、ブルームフィルターの利用、マルチカラムソートの改善版とも言える[Z-Order Optimize](https://docs.databricks.com/delta/optimizations/file-mgmt.html#z-ordering-multi-dimensional-clustering)(多次元クラスタリング)など、さらなるパフォーマンス改善の機能が含まれています。詳細は[こちら](https://docs.databricks.com/delta/optimizations/file-mgmt.html#z-ordering-multi-dimensional-clustering)を参照ください。

# Delta Lake DML: MERGE 

Delta Lakeの`MERGE`コマンドを用いることで、`UPDATE`と`INSERT`の組み合わせである"upsert"を実行できます。upsertを理解するには、既存のテーブル(ターゲットテーブル)と新規レコードと更新レコードを含むソーステーブルを想像します。upsertは以下のように動作します：

- ソーステーブルのレコードがターゲットテーブルの**既存レコードに合致する場合**には、Delta Lakeはレコードを**update**します。
- レコードが**合致しない場合**には、Delta Lakeは新規レコードを**insert**します。

```sql
MERGE INTO events
USING updates
    ON events.eventId = updates.eventId
    WHEN MATCHED THEN UPDATE
        SET events.data = updates.data
    WHEN NOT MATCHED THEN 
        INSERT (date, eventId, data) VALUES (date, eventId, data) 
```

Delta Lakeの`MERGE`コマンドは、Parquetのような他の従来型のデータフォーマットにおいては、複雑かつ重荷となり得るワークフローを劇的に簡素化します。merge/upsertが有用な一般的なシナリオには、チェンジデータキャプチャ、GDPR/CCPAコンプライアンス、セッショニゼーション、レコードの重複排除が含まれます。upsertの詳細に関しては、ブログ記事[Efficient Upserts into Data Lakes with Databricks Delta](https://databricks.com/blog/2019/03/19/efficient-upserts-into-data-lakes-databricks-delta.html)、[Simple, Reliable Upserts and Deletes on Delta Lake Tables using Python API](https://databricks.com/blog/2019/10/03/simple-reliable-upserts-and-deletes-on-delta-lake-tables-using-python-apis.html)、[Schema Evolution in Merge Operations and Operational Metrics in Delta Lake](https://databricks.com/blog/2020/05/19/schema-evolution-in-merge-operations-and-operational-metrics-in-delta-lake.html)を参照ください。

`whenMatched`句による条件の使用を含むプログラム処理など`MERGE`に関するより詳細な情報に関しては、[ドキュメント](https://docs.databricks.com/spark/latest/spark-sql/language-manual/delta-merge-into.html)を参照ください。

## MERGE: 内部処理

Delta Lakeにおける`MERGE`は二つのステップで実行されます:

1. 合致する全てのファイルを選択するために、ターゲットテーブルとソーステーブルの間で**inner join**を実行します。
1. ターゲットテーブルとソーステーブルで選択されたファイルの間で**outer join**を実行し、update/delete/insertされたデータを書き出します。
![](https://databricks.com/wp-content/uploads/2020/09/blog-diving-delta-4.png)

`UPDATE`や`DELETE`の内部処理と異なるのは、Delta　Lakeは`MERGE`を実行するのに*join*を用いる点です。これによって、性能改善方法を探す際にユニークな戦略を取ることができます。

## MERGE: パフォーマンスチューニングのティップス

`MERGE`コマンドの性能を改善するには、mergeを構成する2つのjoinのどちらが性能のリミットとなっているのかを特定する必要があります。

**inner join**がボトルネックである場合(例：Delta Lakeが再度書き込む必要のあるファイルの検索が長時間要する)、以下の戦略をトライしてください。

- 検索空間を狭めるために述語を追加する。
- シャッフルパーティションを調整します。
- ブロードキャストジョインの閾値を調整します。
- テーブル内の小さいファイルが大量にある場合コンパクトにします。しかし、Delta Lakeは再度書き込むファイル全体をコピーする必要があるため、大きすぎるファイルへのコンパクションは行わないでください。

> DatabricksマネージドのDelta Lakeにおいては、更新の局所性を高めるためにZ-orderによる最適化を実行してください。

一方で、**outer join**がボトルネックの場合(例：実際のファイルの再書き込みが長時間要する)、以下の戦略をトライしてください。

- シャッフルパーティションを調整します
    - パーティショニングされたテーブルに対して大量の小さいファイルが生成される場合があります。
    - (Databricks Delta Lakeの最適化された)書き込みの前に自動再パーティショニングを有効化してファイルを削減します。
- ブロードキャスト閾値を調整します。*full outer join*を行う際には、Sparkはブロードキャストジョインを行いません。しかし、*right outer join*を行う際には、Sparkはブロードキャストジョインを行います。必要に応じてブロードキャストの閾値を調整できます。
- ソーステーブル/データフレームをキャッシュします。
    - *ソーステーブル*のキャッシュによって2番目のスキャンをスピードアップしますが、キャッシュの一貫性問題を引き起こす可能性があるので、*ターゲットテーブル*をキャッシュしないよう注意してください。

# まとめ

Delta Lakeは、一般的なビッグデータオペレーションを劇的に簡素化する`UPDATE`、`DELETE`、`MERGE INTO`のDMLコマンドをサポートしています。本記事では、Delta Lakeでこれらのコマンドをどのように使用するのか、それぞれの内部処理、パフォーマンスチューニングのティップスを紹介しました。

# 関連記事
- [Delta Lakeにダイビング #1：トランザクションログを読み解く](https://qiita.com/taka_yayoi/items/b7f628c219463e055592)
- [Delta Lakeにダイビング #2：スキーマの強制、進化](https://qiita.com/taka_yayoi/items/0336b4bbf9b0daaf92d1)
- [Delta Lakeにダイビング #3：DMLの内部処理(Update、Delete、Merge)](https://qiita.com/taka_yayoi/items/c264309e526a1d9ba149)

# その他のリソース
- [Delta Lakeクイックスタートガイド \- Qiita](https://qiita.com/taka_yayoi/items/345f503d5f8177084f24)
- [Databricks documentation on UPDATE, MERGE, and DELETE](https://docs.databricks.com/delta/delta-update.html)
- [Simple, Reliable Upserts and Deletes on Delta Lake Tables using Python APIs](https://databricks.com/blog/2019/10/03/simple-reliable-upserts-and-deletes-on-delta-lake-tables-using-python-apis.html)

### Databricks 無料トライアル

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