Delta LakeとApache Spark™を用いた高信頼データレイクの構築

こちらのノートブックをウォークスルーした内容となっています。

https://databricks.com/notebooks/gallery/IntroductionDeltaLake.html

ノートブックの日本語訳はこちらにあります。

https://github.com/taka-yayoi/public_repo/tree/main/high%20reliable%20data%20lake%20with%20delta%20lake%20and%20spark


**Delta Lake**: Apache Spark™とビッグデータワークロードにACIDトランザクションを提供するオープンソースストレージフォーマット

これは、いくつかのDelta Lakeの機能を説明する入門ノートブックです。

* **オープンフォーマット**: blobストレージ上のParquetフォーマットとして格納されます。
* **ACIDトランザクション**: 複雑かつ同時実行されるデータパイプラインでデータの完全性と読み取りの一貫性を保証します。
* **スキーマ強制、スキーマ進化**: 予期しない書き込みをブロックすることでデータの綺麗さを保持します。
* **監査履歴**: テーブルに生じた全てのオペレーションの履歴。
* **タイムトラベル**: 時間あるいはバージョン番号による以前のバージョンのテーブルへのクエリー。
* **deleteとupsert**: プログラミングAPIによるテーブルのdeleteとupsert(update + insert)のサポート。
* **スケーラブルなメタデータ管理**: Sparkを用いることでメタデータのオペレーションをスケールさせ、数百万のファイルを取り扱うことができます。
* **バッチ、ストリーミングソース、ストリーミングシンクの統合**: Delta Lakeのテーブルはバッチテーブルでもあり、ストリーミングのソースとシンクでもあります。ストリーミングデータの取り込み、バッチによる過去のバックフィル、インタラクティブなクエリーすべてをすぐに活用することができます。

このノートブックは[SAIS EU 2019 Delta Lake Tutorial](https://github.com/delta-io/delta/tree/master/examples/tutorials/saiseu19)の修正バージョンです。使用データは、[Lending Club](https://www.kaggle.com/wendykan/lending-club-loan-data)の公開データの修正バージョンです。2012年から2017年の間にファンディングされた全てのローンが含まれています。それぞれのローンには、申請者の情報、現在のローンのステータス(Current, Late, Fully Paid, etc.)、最新の支払い情報が含まれています。データの完全なビューに関しては、[こちら](https://resources.lendingclub.com/LCDataDictionary.xlsx)のデータ辞書を参照してください。

# Delta Lakeテーブルへのデータのロード

最初にデータを読み込み、Delta Lakeテーブルとして保存しましょう。

```py:Python
import re
from pyspark.sql.types import * 

# Username を取得
username_raw = dbutils.notebook.entry_point.getDbutils().notebook().getContext().tags().apply('user')
# Username の英数字以外を除去し、全て小文字化
username = re.sub('[^A-Za-z0-9]+', '', username_raw).lower()

# ファイル格納パス
work_path = f"dbfs:/tmp/databricks_handson/{username}/delta_introduction"

# パスを表示
print(f"path_name: {work_path}")
```

```py:Python
spark.sql("set spark.sql.shuffle.partitions = 1")

# 読み込むデータ
sourcePath = "/databricks-datasets/learning-spark-v2/loans/loan-risks.snappy.parquet"

# Delta Lakeのパスの設定
deltaPath = f"{work_path}/loans_delta"

# フォルダーが存在する場合には削除
dbutils.fs.rm(deltaPath, recurse=True)

# 同じローンデータを用いたDeltaテーブルの作成
(spark.read.format("parquet").load(sourcePath) 
  .write.format("delta").save(deltaPath))

spark.read.format("delta").load(deltaPath).createOrReplaceTempView("loans_delta")
print("Defined view 'loans_delta'")
```

# データの探索

データを探索してみましょう。

```py:Python
spark.sql("SELECT count(*) FROM loans_delta").show()
```
![Screen Shot 2022-07-22 at 13.08.14.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/5539fa1c-363b-0394-6f3e-1b46da245721.png)

```py:Python
display(spark.sql("SELECT * FROM loans_delta LIMIT 5"))
```
![Screen Shot 2022-07-22 at 13.08.58.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/b26c36b7-1228-0948-7e2c-01e305fb5dae.png)

マジックコマンド`%sql`を用いても同様の処理が可能です。

```sql:SQL
%sql
SELECT * FROM loans_delta LIMIT 5
```
![Screen Shot 2022-07-22 at 13.09.50.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f40eacd7-f097-fd5d-125e-24fd53ee54c8.png)

# Delta Lakeテーブルへのデータストリームのロード

Deltaテーブルを作成することのメリットの一つに、「バッチデータ処理だけでなくストリーミング処理」でも利用できるというものがあります。具体的にはストリーミングのソースやシンクとしてDeltaテーブルを使用することができます。

ここでは、ランダムに生成されるローンIDと金額によるデータストリームを作成します。さらに、有用なユーティリティ関数をいくつか定義します。

```py:Python
import random
import os
from pyspark.sql.functions import *
from pyspark.sql.types import *

def random_checkpoint_dir(): 
  return f"{work_path}/tmp/chkpt/%s" % str(random.randint(0, 10000))


# 州をランダムに生成するユーザー定義関数
states = ["CA", "TX", "NY", "WA"]

@udf(returnType=StringType())
def random_state():
  return str(random.choice(states))

# ランダムに生成されたデータのストリームによるストリーミングクエリーを起動し、Deltaテーブルにデータを追加する関数
def generate_and_append_data_stream():

  newLoanStreamDF = (spark.readStream.format("rate").option("rowsPerSecond", 5).load() 
    .withColumn("loan_id", 10000 + col("value")) 
    .withColumn("funded_amnt", (rand() * 5000 + 5000).cast("integer")) 
    .withColumn("paid_amnt", col("funded_amnt") - (rand() * 2000)) 
    .withColumn("addr_state", random_state())
    .select("loan_id", "funded_amnt", "paid_amnt", "addr_state"))
    
  checkpointDir = f"{work_path}/chkpt"

  streamingQuery = (newLoanStreamDF.writeStream 
    .format("delta") 
    .option("checkpointLocation", checkpointDir) 
    .trigger(processingTime = "10 seconds") 
    .start(deltaPath))

  return streamingQuery

# 全てのストリーミングクエリーを停止する関数 
def stop_all_streams():
  # 全てのストリームを停止
  print("Stopping all streams")
  for s in spark.streams.active:
    s.stop()
  print("Stopped all streams")
  print("Deleting checkpoints")  
  dbutils.fs.rm(f"{work_path}/chkpt", True)
  print("Deleted checkpoints")
```

以下のコマンドを実行してストリームを起動します。

```py:Python
streamingQuery = generate_and_append_data_stream()
```

ストリーミングクエリーが起動すると専用のダッシュボードが表示されます。こちらのダッシュボードを用いることで、リアルタイムにストリーミングの流量などを確認することができます。
![Screen Shot 2022-07-22 at 13.12.43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/65dead6b-2bcd-0958-4f44-94e5f08a22cf.png)

このケースでは、Deltaテーブルがストリーミングのシンクになっています。通常のテーブルと同様にテーブルに対してクエリーを実行することができます。

テーブルのレコード数をカウントすることで、ストリーミングクエリーがテーブルにデータが追加されていることを確認することができます。以下のセルを複数回実行してください。実行の都度カウントアップしていく様子を確認することができます。

```py:Python
display(spark.sql("SELECT count(*) FROM loans_delta"))
```

**処理が不要になったら、全てのストリーミングクエリーを停止することを忘れないでください。**

:::note warn
**重要！**
ストリーミング処理は明示的に停止しない限り処理を続行します。クラスターの自動停止も効かないのでコスト増に注意してください。
:::

```py:Python
stop_all_streams()
```

# データ破損を防ぐために書き込み時のスキーマを強制

ローンが停止されたかどうかを示す追加のカラム`closed`を含むデータを書き込むことで、この機能をテストしてみましょう。このカラムはテーブルには存在しないことに注意してください。

```py:Python
cols = ['loan_id', 'funded_amnt', 'paid_amnt', 'addr_state', 'closed']

items = [
  (1111111, 1000, 1000.0, 'TX', True), 
  (2222222, 2000, 0.0, 'CA', False)
]

from pyspark.sql.functions import *

loanUpdates = (spark
                .createDataFrame(items, cols)
                .withColumn("funded_amnt", col("funded_amnt").cast("int")))
```

```py:Python
# 以下の行のコメントを解除して実行ししてください。エラーになるはずです。
loanUpdates.write.format("delta").mode("append").save(deltaPath)
```
![Screen Shot 2022-07-22 at 13.20.51.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/fd3181a7-a3ad-f014-cee5-28bb7938dfd4.png)

# 変化するデータに対応するためにスキーマを進化

スキーマ強制は意図しないスキーマの汚染を防御するためのものですが、ビジネス要件の変化に合わせて明示的にスキーマを変更しなくてはならないケースがあります。この場合、オプション`mergeSchema`に`true`を設定します。

```py:Python
(loanUpdates.write.format("delta").mode("append")
  .option("mergeSchema", "true")
  .save(deltaPath))
```

スキーマを確認するために再度テーブルにクエリーを実行してみましょう。

```py:Python
display(spark.read.format("delta").load(deltaPath).filter("loan_id = 1111111"))
```
![Screen Shot 2022-07-22 at 13.22.16.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/8ddb05cd-27c3-5dfc-7a92-b049e7fbc144.png)

既存のレコードを読み込んだ際、新規のカラムはNULLと見做されます。

```py:Python
display(spark.read.format("delta").load(deltaPath))
```
![Screen Shot 2022-07-22 at 13.22.52.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/60867103-9971-cb1a-b4f4-ab0866d861f6.png)

# 既存データの変換

既存データをどのように変換できるのかをみていきましょう。でも最初に、スキーマが変化したのでビューが変更に追従できるように、ビューを再定義することでテーブルに対するビューを更新しましょう。

```py:Python
spark.read.format("delta").load(deltaPath).createOrReplaceTempView("loans_delta")
print("Defined view 'loans_delta'")
```

## エラーを修正するためにローンデータを更新

* データをレビューしてみると、`addr_state = 'OR'`に割り当てられた全てのローンは`addr_state = 'WA'`に割り当てられるべきであることがわかりました。
* Parquetで`update`を行うには以下のことが必要となります。
  * 更新されない全ての行を新規テーブルにコピー
  * 更新される全ての行をデータフレームにコピーし、データを更新
  * 新規テーブルに上述のデータフレームをinsert
  * 古いテーブルを削除
  * 新規テーブルの名称を古いものに更新

Deltaテーブルであれば、通常のテーブル同様に`UPDATE`を行うことができます。

```py:Python
display(spark.sql("""SELECT addr_state, count(1) FROM loans_delta WHERE addr_state IN ('OR', 'WA', 'CA', 'TX', 'NY') GROUP BY addr_state"""))
```
![Screen Shot 2022-07-22 at 13.24.45.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f32f16c8-b604-f52e-8311-e4fd5b4a410b.png)

データを修正しましょう。

```py:Python
from delta.tables import *

deltaTable = DeltaTable.forPath(spark, deltaPath)
deltaTable.update("addr_state = 'OR'",  {"addr_state": "'WA'"})
```

もう一度データを確認しましょう。

```py:Python
display(spark.sql("""SELECT addr_state, count(1) FROM loans_delta WHERE addr_state IN ('OR', 'WA', 'CA', 'TX', 'NY') GROUP BY addr_state"""))
```
![Screen Shot 2022-07-22 at 13.25.48.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f7e7b479-3848-41c4-ba31-0b120aabb5d2.png)

データが更新されていることがわかります。

## General Data Protection Regulation (GDPR)に基づいてテーブルからユーザーデータを削除

Delta Lakeテーブルから述語にマッチするデータを削除することができます。ここでは、完済したローンすべてを削除したいものとします。最初にどれだけレコードがあるのかを確認しましょう。

```py:Python
display(spark.sql("SELECT COUNT(*) FROM loans_delta WHERE funded_amnt = paid_amnt"))
```
![Screen Shot 2022-07-22 at 13.26.40.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/0c0b871a-c635-ef39-26e5-d6b0c62ce8d0.png)

Parquetの場合、部分的な削除ができないためデータの総入れ替えを行わなくてはなりませんが、Deltaであればこれも通常のテーブル同様に`DELETE`を行うことができます。

```py:Python
from delta.tables import *

deltaTable = DeltaTable.forPath(spark, deltaPath)
deltaTable.delete("funded_amnt = paid_amnt")
```

完済したローンの数をチェックしてみます。

```py:Python
display(spark.sql("SELECT COUNT(*) FROM loans_delta WHERE funded_amnt = paid_amnt"))
```
![Screen Shot 2022-07-22 at 13.28.14.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/ffc5497a-84e2-e638-9875-0591e2d11a46.png)

問題なく削除できていることがわかります。

## mergeを用いた変更データのテーブルへのupsert

一般的なユースケースは、あるOLAPテーブルでなされた行の変更を、OLAPワークロードの別のテーブルに複製するチェンジデータキャプチャ(CDC)です。ここでのローンデータのサンプルで使ってみるために、新たなローンや既存のローンに対する更新情報を含む別のテーブルがあるものとします。さらに、この変更テーブルは`loan_delta`テーブルと同じスキーマとなっています。SQLコマンド`MERGE`をベースとした`DeltaTable.merge()`を用いることで、これらの変更をテーブルにupsertすることができます。

```py:Python
display(spark.sql("select * from loans_delta where addr_state = 'NY' and loan_id < 30"))
```

![Screen Shot 2022-07-22 at 13.29.05.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d61ab577-3faf-87b9-798f-e487bc982c4b.png)

このデータにはいくつかの変更があり、ある1つのローンは完済され、別の新たなローンが追加されたものとします。

```py:Python
cols = ['loan_id', 'funded_amnt', 'paid_amnt', 'addr_state', 'closed']

items = [
  (11, 1000, 1000.0, 'NY', True),   # ローンの完済
  (12, 1000, 0.0, 'NY', False)      # 新たなローン
]

loanUpdates = spark.createDataFrame(items, cols)
```

次に、`merge`オペレーションを用いて変更データでテーブルを更新しましょう。

```py:Python
from delta.tables import *

deltaTable = DeltaTable.forPath(spark, deltaPath)

(deltaTable
  .alias("t")
  .merge(loanUpdates.alias("s"), "t.loan_id = s.loan_id") 
  .whenMatchedUpdateAll() 
  .whenNotMatchedInsertAll() 
  .execute())
```

テーブルが更新されたことを確認しましょう。

```py:Python
display(spark.sql("select * from loans_delta where addr_state = 'NY' and loan_id < 30"))
```

![Screen Shot 2022-07-22 at 13.31.18.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/94dc0483-a514-dd3a-7980-74ac872feb79.png)

loan_idを用いて照合が行われ、行が存在する場合には更新され、存在しない場合にはインサートされていることがわかります。

# オペレーション履歴を用いたデータ変更の監査

ここまで行なったDeltaテーブルに対する全ての変更は、テーブルのトランザクションログのコミットとして記録されます。Deltaテーブルやディレクトリに書き込みを行うと、全てのオペレーションは自動的にバージョン管理されます。テーブルの履歴を参照するために`HISTORY`コマンドを使用することができます。

```py:Python
from delta.tables import *

deltaTable = DeltaTable.forPath(spark, deltaPath)
display(deltaTable.history())
```
![Screen Shot 2022-07-22 at 13.32.59.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/6be64921-cbd4-6d5c-e905-b7deec6bc70b.png)

誰がいつどのような操作を行ったのかが記録されるので、監査ログとして用いることも可能です。

履歴に対してクエリーを実行することもできます。

```py:Python
display(deltaTable.history(4).select("version", "timestamp", "operation", "operationParameters"))
```
![Screen Shot 2022-07-22 at 13.34.07.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/25c6596c-b637-e314-9a15-3a997a0b5429.png)

# タイムトラベルを用いたテーブルの過去のスナップショットへのクエリー

Delta Lakeのタイムトラベル機能を用いることで、テーブルの以前のバージョンにアクセスすることができます。この機能を活用できるユースケースには以下のようなものがあります。

* データ変更の監査
* 実験 & レポートの再現
* ロールバック

Python、Scala、SQL文法を用いてタイムスタンプ、バージョン番号を用いてクエリーを行うことができます。この例では、Python文法を用いて特定のバージョンにクエリーを行います。

詳細に関しては、[Introducing Delta Time Travel for Large Scale Data Lakes](https://databricks.com/blog/2019/02/04/introducing-delta-time-travel-for-large-scale-data-lakes.html)や[ドキュメント](https://docs.delta.io/latest/delta-batch.html#deltatimetravel)をご覧ください。

**完済ローンを含んでいるデータの削除を行う前のテーブルをクエリーしてみましょう。**

```py:Python
previousVersion = deltaTable.history(1).select("version").first()[0] - 3

(spark.read.format("delta")
  .option("versionAsOf", previousVersion)
  .load(deltaPath)
  .createOrReplaceTempView("loans_delta_pre_delete"))

display(spark.sql("SELECT COUNT(*) FROM loans_delta_pre_delete WHERE funded_amnt = paid_amnt"))
```

削除前にはあった完済ローンと同じ数を確認することができましたので、削除前の状態を参照していることがわかります。
![Screen Shot 2022-07-22 at 13.26.40.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/0c0b871a-c635-ef39-26e5-d6b0c62ce8d0.png)

このように、オブジェクトストレージ上のデータをDelta Lakeで管理し、Sparkでアクセすることで、データレイク上のデータの品質を保つためのさまざまな機能を活用できるようになります。

### Databricks 無料トライアル

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