Delta LakeによるSparkクエリーの高速化

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

https://docs.databricks.com/delta/optimizations/optimization-examples.html#delta-lake-on-databricks-optimizations-python-notebook

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

https://github.com/taka-yayoi/public_repo/blob/main/delta_optimize/Databricks%20Delta%E3%81%AB%E3%82%88%E3%82%8B%E9%AB%98%E6%80%A7%E8%83%BDSpark%E3%82%AF%E3%82%A8%E3%83%AA%E3%83%BC%20(Python).py

:::note warn
**重要！**
こちらの例では、i3.xlarge(メモリ30.5GB、4コア)のシングルノードのクラスターを使用しています。こちらで示している性能値は参考値と捉えてください。
:::

# Delta Lakeとは

[Delta Lake](https://delta.io/)はデータレイク上で高性能のデータ処理、データ品質の確保を可能にするストレージレイヤーソフトウェアです。[Apache Parquet](https://databricks.com/jp/glossary/what-is-parquet)をベースとしたオープンなデータフォーマットを提供します。

データレイクの概要、歴史、課題に関してはこちらを参照ください。

https://databricks.com/jp/discover/data-lakes/introduction

> データレイクとは、膨大な量のデータを集約し、未加工のネイティブな形式で格納するレポジトリです。ファイルやフォルダ内にデータを格納する階層型データウェアハウスと比較して、データレイクはフラットなアーキテクチャとオブジェクトストレージでデータを格納します。データをメタデータタグと一意の識別子で格納するオブジェクトストレージにより、リージョン間でのデータの検索と取得は容易で、性能を向上させます。安価なオブジェクトストレージとオープンフォーマットを活用し、データレイクは、多くのアプリケーションにおけるデータ活用を可能にします。

Delta Lakeは以下のような機能を提供することで、高性能・高信頼のデータレイクを実現します。

- Sparkにおける[ACIDトランザクション](https://qiita.com/taka_yayoi/items/94db8363720687d17d6e): シリアライズ可能なアイソレーションレベルによって、読み取り側で一貫性のないデータを参照することがなくなります。
- スケーラブルなメタデータのハンドリング: 数十億のファイルから構成されるパタバイト規模の全てのメタデータを容易に取り扱うことができる、Sparkの分散処理のパワーを活用します。
- [ストリーミング](https://qiita.com/taka_yayoi/items/41d97a7c552f947729a0)と[バッチ](https://qiita.com/taka_yayoi/items/47b58eb8696b278a3a82)の統合: Delta Lakeのテーブルはバッチテーブルであるのと同時に、ストリーミングのソース、シンクでもあります。ストリーミングのデータ取り込み、バッチによる過去データのバックフィル、インタラクティブなクエリーすべてを簡単に実行できます。
- スキーマ強制: データ取り込みの際に不正なレコードの挿入を防ぐために、自動でスキーマの変更をハンドリングします。
- [タイムトラベル](https://qiita.com/taka_yayoi/items/47b58eb8696b278a3a82#%E9%81%8E%E5%8E%BB%E3%81%AE%E3%83%86%E3%83%BC%E3%83%96%E3%83%AB%E3%81%AE%E3%82%B9%E3%83%8A%E3%83%83%E3%83%97%E3%82%B7%E3%83%A7%E3%83%83%E3%83%88%E3%81%AE%E3%82%AF%E3%82%A8%E3%83%AA%E3%83%BC%E3%82%BF%E3%82%A4%E3%83%A0%E3%83%88%E3%83%A9%E3%83%99%E3%83%AB): データのバージョン管理によって、ロールバック、完全な履歴の監査証跡、再現可能な機械学習エクスペリメントを実現します。
- [upsert](https://qiita.com/taka_yayoi/items/0d8350a2b4739392a52f#merge%E3%82%92%E7%94%A8%E3%81%84%E3%81%9F%E3%83%86%E3%83%BC%E3%83%96%E3%83%AB%E3%81%B8%E3%81%AEupsert)と[delete](https://qiita.com/taka_yayoi/items/0d8350a2b4739392a52f#%E3%83%86%E3%83%BC%E3%83%96%E3%83%AB%E3%81%8B%E3%82%89%E3%81%AEdelete): チェンジデータキャプチャやSCD(slowly-changing-operation)オペレーション、ストリーミングによるupsertなど複雑なユースケースを実現するためのmerge、update、deleteオペレーションをサポートします。

# Sparkとは

ビッグデータと機械学習ワークロードに最適化された、非常に高速な分散処理フレームワークです。

https://databricks.com/jp/spark/about

# SparkとDelta Lakeと組み合わせることのメリット

クラウドのオブジェクトストレージに格納されているさまざまな種類のデータを高速かつ信頼性を持って処理できるようになります。

**大容量データ・多種多様なデータを高速に処理**

- 容量の制限がないのでデータ量の増加にも柔軟に対応できます。
- Delta Lakeには画像、テキストなど非構造化データを含めることができるので、データの種類を問わずにデータを格納するとができます。
- SparkはネイティブでDelta Lakeのデータを読み書きできるので、大量データを簡単に並列処理することができます。

**データの品質保証**

- ACIDトランザクションが保証されるので、データ書き込みジョブによるデータ破損を回避できます。
- Delta Lakeのスキーマ強制によって、Sparkで取り扱うデータのスキーマを固定、あるいは必要に応じて進化させることができます。

# ノートブックのウォークスルー

この例では、どのようにDelta Lakeがクエリー性能を最適化するのかを見ていきます。Parquetフォーマットを用いた標準的なテーブルを作成し、レーテンシーを観察するために簡単なクエリーを実行します。次に、標準的なテーブルとDelta Lakeテーブルの間での性能の違いを見るために、同じテーブルのDelta Lakeバージョンに対して二つ目のクエリーを実行します。

シンプルに以下の4ステップを踏んでいきます。

* __Step 1__ : USのフライトスケジュールデータを用いて標準的なParquetベーステーブルを作成します。
* __Step 2__ : 年間を通じて出発空港、月毎のフライト数を計算するクエリーを実行します。
* __Step 3__ : Delta Lakeを用いてフライトテーブルを作成し、テーブルの最適化を行います。
* __Step 4__ : ステップ２のクエリーを再実行し、レーテンシーを観察します。

## 環境設定

こちらでは、他のユーザーとファイルパス、データベースが重複しないようにユーザー名を含むファイルパス、データベースを準備します。

```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()

# ユーザー固有のデータベース名を生成します
db_name = f"databricks_handson_{username}"

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

# データベースの準備
spark.sql(f"DROP DATABASE IF EXISTS {db_name} CASCADE")
spark.sql(f"CREATE DATABASE IF NOT EXISTS {db_name}")
spark.sql(f"USE {db_name}")

# データベースを表示。
print(f"database_name: {db_name}")
print(f"path_name: {work_path}")
```

![Screen Shot 2022-07-06 at 14.37.59.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/435be83a-c1d4-0652-52e0-aabad5881655.png)

## ステップ0: フライトデータの読み込み

```py:Python
flights = spark.read.format("csv") \
  .option("header", "true") \
  .option("inferSchema", "true") \
  .load("/databricks-datasets/asa/airlines/2008.csv")
```

```py:Python
flights.count()
```

![Screen Shot 2022-07-06 at 14.39.21.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/b4e684ef-b163-068c-195e-83a43b84c0c6.png)

約700万レコード、700MBのデータとなっています。

## ステップ1: フライトデータを用いたParquetテーブルの書き込み

カラム`Origin`でParquetのパーティション(Sparkのパーティションとは異なります)を作成し、ディスクに書き込みます。

```py:Python
flights.write.format("parquet").mode("overwrite").partitionBy("Origin").save(f"{work_path}/flights_parquet")
```

![Screen Shot 2022-07-06 at 14.40.11.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/eb4e7c6d-732c-aa9f-d8d5-d4f5688eab86.png)

ステップ1が完了すると、"flights"テーブルには年を通じたUSのフライト詳細が含まれます。

次にステップ2では、週の初日のフライト数の月間合計に基づくトップ20の都市を取得するクエリーを実行します。

## ステップ2: クエリー実行

```py:Python
from pyspark.sql.functions import count

flights_parquet = spark.read.format("parquet").load(f"{work_path}/flights_parquet")

display(flights_parquet.filter("DayOfWeek = 1").groupBy("Month","Origin").agg(count("*").alias("TotalFlights")).orderBy("TotalFlights", ascending=False).limit(20))
```

![Screen Shot 2022-07-06 at 14.41.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/5b3c92d6-1003-9cf5-2ca7-6d02526d5a07.png)

ステップ2が完了すると、標準的な"flights_parquet"テーブルにおけるレーテンシーを観測することができます。

ステップ3とステップ4においては、Deltaテーブルで同じことを行います。今回はクエリーを実行する前に、検索を高速化するためにデータを最適化するために`OPTIMIZE`と`ZORDER`を実行します。

## ステップ3: フライトデータを用いたDeltaテーブルの書き込み

```py:Python
flights.write.format("delta").mode("overwrite").partitionBy("Origin").save(f"{work_path}/flights_delta")
```

![Screen Shot 2022-07-06 at 14.42.10.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c32d6697-64f9-4c96-4083-0f407086e443.png)

## ステップ3(続き): Databricks DeltaテーブルのOPTIMIZE

Z-Orderingは関連する情報を同じファイルセットに配置するパフォーマンスチューニングの[テクニック](https://en.wikipedia.org/wiki/Z-order_curve)です。Delta Lakeのデータスキッピングアルゴリズムは読み取る必要があるデータ量を劇的に削減するために、この局所性(locality)を自動で活用します。データをZ-Orderするには、`ZORDER BY`句に並び替えを行うカラムを指定します。

**参考資料**: [Databricksにおけるデータファイル管理によるパフォーマンスの最適化 \- Qiita](https://qiita.com/taka_yayoi/items/a82b0b8fbdc74e6f9f01#z-ordering%E5%A4%9A%E6%AC%A1%E5%85%83%E3%82%AF%E3%83%A9%E3%82%B9%E3%82%BF%E3%83%AA%E3%83%B3%E3%82%B0)

```py:Python
display(spark.sql("DROP TABLE IF EXISTS flights"))

display(spark.sql(f"CREATE TABLE flights USING DELTA LOCATION '{work_path}/flights_delta'"))
                  
display(spark.sql("OPTIMIZE flights ZORDER BY (DayofWeek)"))
```

![Screen Shot 2022-07-06 at 14.43.19.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4fdcd767-2528-5725-a456-ffda0096d557.png)

## ステップ4: ステップ2のクエリーを実行してレーテンシーを測定

```py:Python
flights_delta = spark.read.format("delta").load(f"{work_path}/flights_delta")

display(flights_delta.filter("DayOfWeek = 1").groupBy("Month","Origin").agg(count("*").alias("TotalFlights")).orderBy("TotalFlights", ascending=False).limit(20))
```

![Screen Shot 2022-07-06 at 14.44.09.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/1c0003e8-c4ad-a228-8e9b-543b40ad81c6.png)

`OPTIMIZE`の実行後は、Deltaテーブルに対するクエリーが非常に高速になりました。クエリーをどのくらい高速化できるかは、処理を実行するクラスターに依存しますが、標準的なテーブルと比較して**5-10倍**の高速化を実現することができます。

今回の例では`OPTIMIZE`を行うことで、クエリーの時間は`1.59分`から`17秒`になり、**約5.6倍**高速になっています。

### Databricks 無料トライアル

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