Oracle Cloud Infrastructure (OCI) Data Flow で Delta Lake を試してみた 


# はじめに

Oracle Cloud Infrastructure (OCI) Data Flow で Delta Lake がサポートされました。
ということで、基本的な動作を試してみたいと思います。

OCI Data Flow のドキュメントは [**こちら**](https://docs.oracle.com/ja-jp/iaas/data-flow/using/delta-lake-about.htm)

# Delta Lake とは

[**Delta Lake のサイト**](https://delta.io/ ) では、「Lakehouse アーキテクチャを構築できるオープンソース」だと書いてありますが、これではちょっと分からない...。
Delta Lake を使うと、オブジェクトストレージにあるデータソースに対してトランザクショナルな更新処理を行うようなユースケースに対応できます。また、タイムトラベル機能によって更新処理毎に都度スナップショットが記録されるため、タイムスタンプやバージョン番号で任意の時点のスナップショットを取得できます。素の Spark だと、書き込み処理のモードは、"overwrite" か "append" かのどちらかですが、Delta Lake を使うと特定の行を挿入・更新・削除するなどもう少し賢い更新処理が可能です。さらに、更新処理は ACID に行われるので、複数の更新が同時に発生した場合でもデータの一貫性が保たれます。その他、Delta Lake にはスキーマを検証して乱雑な書き込みを防ぎデータの品質を保つ機能などが備わっています。

# OCI Data Flow で Delta Lake を使う


OCI Data Flow の Spark 3.2.1 処理エンジンと統合されているため、アプリケーション側で追加の Spark 構成を設定する必要はありません。Sparkのバージョンは 3.2.1 を選択して下さい (2022年11月現在)。

## Python (pyspark)

ローカルで実行する訳ではないので、Delta Lake のライブラリが必須ではないですが、IDE での開発効率を高めるために pip でインストールしておきます。

```bash
pip install delta-spark
```
SparkSession の作成はいつも通りのやり方で OK です。

```python
spark = SparkSession.builder.appName("Delta Lake Test").getOrCreate()
```

## Scala

build.sbt に Delta Lake のライブラリを追加してコンパイルできるようにします。

```
libraryDependencies += "io.delta" %% "delta-core" % "2.1.1" % "provided"
```

こちらも SparkSession の作成はいつも通りのやり方で OK です。

```scala
val spark = SparkSession.builder().appName("Delta Lake Test").getOrCreate()
```

# 動作確認をする

今回使ったソース、Python (pyspark)版 と Scala 版はこちらに。

https://gist.github.com/tkote/65c181bfc2a47a7da270103fe1437512

https://gist.github.com/tkote/d28a99d2e491b814d7ceb63a2d1a8ff6

ここでは、pyspark 版で動作確認を見ていきます。

まずはオブジェクト・ストレージのパス (oci://～で始まる文字列) を引数から受取って、Delta Lake のファイルを書き出します。


```python
deltaTablePath = sys.argv[1] # URL of Object Storage - oci://~
print(f">>> deltaTablePath: {deltaTablePath}")

print(">>> Create table with id in range(0, 5)")
spark.range(0, 5).withColumn("data", lit("create") ) \
  .write.format("delta").mode("overwrite").save(deltaTablePath)
spark.read.format("delta").load(deltaTablePath).show()
```

出力

```
>>> Create table with id in range(0, 5)
+---+------+
| id|  data|
+---+------+
|  2|create|
|  3|create|
|  4|create|
|  0|create|
|  1|create|
+---+------+
```

さらに 5行分データを追加します。

```python
print(">>> Append rows with id in range(5, 10)")
spark.range(5, 10).withColumn("data", lit("append")) \
  .write.format("delta").mode("append").save(deltaTablePath)
spark.read.format("delta").load(deltaTablePath).show()
```

出力

```
>>> Append rows with id in range(5, 10)
+---+------+
| id|  data|
+---+------+
|  2|create|
|  3|create|
|  4|create|
|  7|append|
|  8|append|
|  9|append|
|  0|create|
|  1|create|
|  5|append|
|  6|append|
+---+------+
```

id が偶数の行のデータを更新してみます。

```python
deltaTable = DeltaTable.forPath(spark, deltaTablePath)

print(">>> Update rows where id is even")
deltaTable.update(condition = expr("id % 2 == 0"), set = { "data" : lit("update") })
deltaTable.toDF().show()
```

出力

```
>>> Update rows where id is even
+---+------+
| id|  data|
+---+------+
|  2|update|
|  3|create|
|  4|update|
|  7|append|
|  8|update|
|  9|append|
|  0|update|
|  1|create|
|  5|append|
|  6|update|
+---+------+
```

さらに、id が 3 の倍数の行を削除します。

```python
print(">>> Delete rows having id which remainder divided by 3 is 0")
deltaTable.delete(condition = expr("id % 3 == 0"))
deltaTable.toDF().show()
```

出力

```
>>> Delete rows having id which remainder divided by 3 is 0
+---+------+
| id|  data|
+---+------+
|  2|update|
|  4|update|
|  7|append|
|  8|update|
|  1|create|
|  5|append|
+---+------+
```

新しい DataFrame を作って、マージします。

```python
print(">>> Upsert (merge) new data with id in range(0, 20)")
newData = spark.range(0, 20).withColumn("data", lit("upsert") )
deltaTable.alias("oldData") \
  .merge(newData.alias("newData"), "oldData.id = newData.id") \
  .whenMatchedUpdate(set = { "data" : concat(col("oldData.data"), lit("/merge")) }) \
  .whenNotMatchedInsert(values = { "id" : newData.id, "data" : lit("/merge") }) \
  .execute()
deltaTable.toDF().show()
```

出力

```
>>> Upsert (merge) new data with id in range(0, 20)
+---+------------+
| id|        data|
+---+------------+
|  0|      /merge|
|  1|create/merge|
|  2|update/merge|
|  3|      /merge|
|  4|update/merge|
|  5|append/merge|
|  6|      /merge|
|  7|append/merge|
|  8|update/merge|
|  9|      /merge|
| 10|      /merge|
| 11|      /merge|
| 12|      /merge|
| 13|      /merge|
| 14|      /merge|
| 15|      /merge|
| 16|      /merge|
| 17|      /merge|
| 18|      /merge|
| 19|      /merge|
+---+------------+
```

全件削除してみます。

```python
print(">>> Delete all rows")
deltaTable.delete(col("id") >= 0)
deltaTable.toDF().show()
```

出力

```
>>> Delete all rows
+---+----+
| id|data|
+---+----+
+---+----+
```

ここまでの更新履歴を確認します。

```python
print(">>> Show history")
deltaTable.history().show()
```

出力

```
>>> Show history
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+--------------+-------------+--------------------+------------+--------------------+
|version|          timestamp|userId|userName|operation| operationParameters| job|notebook|clusterId|readVersion|isolationLevel|isBlindAppend|    operationMetrics|userMetadata|          engineInfo|
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+--------------+-------------+--------------------+------------+--------------------+
|      5|2022-10-28 02:33:08|  null|    null|   DELETE|{predicate -> ["(...|null|    null|     null|          4|  Serializable|        false|{numRemovedFiles ...|        null|Apache-Spark/3.2....|
|      4|2022-10-28 02:32:58|  null|    null|    MERGE|{predicate -> (ol...|null|    null|     null|          3|  Serializable|        false|{numTargetRowsCop...|        null|Apache-Spark/3.2....|
|      3|2022-10-28 02:32:49|  null|    null|   DELETE|{predicate -> ["(...|null|    null|     null|          2|  Serializable|        false|{numRemovedFiles ...|        null|Apache-Spark/3.2....|
|      2|2022-10-28 02:32:39|  null|    null|   UPDATE|{predicate -> ((i...|null|    null|     null|          1|  Serializable|        false|{numRemovedFiles ...|        null|Apache-Spark/3.2....|
|      1|2022-10-28 02:32:22|  null|    null|    WRITE|{mode -> Append, ...|null|    null|     null|          0|  Serializable|         true|{numFiles -> 2, n...|        null|Apache-Spark/3.2....|
|      0|2022-10-28 02:32:06|  null|    null|    WRITE|{mode -> Overwrit...|null|    null|     null|       null|  Serializable|        false|{numFiles -> 2, n...|        null|Apache-Spark/3.2....|
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+--------------+-------------+--------------------+------------+--------------------+
```

バージョン 4 のデータを取り出します。

```python
print(">>> Show version 4")
spark.read.format("delta").option("versionAsOf", 4).load(deltaTablePath).show()
```

出力

```
>>> Show version 4
+---+------------+
| id|        data|
+---+------------+
|  0|      /merge|
|  1|create/merge|
|  2|update/merge|
|  3|      /merge|
|  4|update/merge|
|  5|append/merge|
|  6|      /merge|
|  7|append/merge|
|  8|update/merge|
|  9|      /merge|
| 10|      /merge|
| 11|      /merge|
| 12|      /merge|
| 13|      /merge|
| 14|      /merge|
| 15|      /merge|
| 16|      /merge|
| 17|      /merge|
| 18|      /merge|
| 19|      /merge|
+---+------------+
```

最後に、この状態でオブジェクト・ストレージがどうなっているか見てみましょう。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544382/d350779b-0322-046b-676d-012be1eae8ec.png)

`_delta_log` ディレクトリ内の json ファイルでメタデータが管理されています。


# まとめ

OCI Data Flow で Delta Lake を試してみました。アプリケーション側に特別な設定等は必要なく直ぐに Delta Lake の機能を利用できます。従来単純にオブジェクト・ストレージから読み込んだり逆に書き出したりするだけでしたが、Delta Lake により更新履歴付きの更新処理が可能になったことで、ぐんとアプリケーションの幅が広がると思います。

## 補足

今回は OCI Data Flow から Delta Lake を使ってみましたが、もちろん普通に 素の Spark から使うこともできます。OCI のオブジェクト・ストレージを使って Delta Lake のファイルを操作するための[**設定方法**](https://docs.delta.io/latest/delta-storage.html#oracle-cloud-infrastructure)も Delta Lake のドキュメントの一部として提供されています。