Databricks（Spark）にてDelta Lake形式とParquet形式における実行計画のデータサイズの比較

## 概要
Databricks（Spark）にてDelta Lake形式とParquet形式における実行計画のデータサイズを比較したところ、Delta Lake形式では約1万分の1のサイズとなりました。Delta Lake形式の利用により、Parquet形式を利用よりも性能の向上が期待できます。

パーティション列を設定したParquet形式のデータセットとDelta Lake形式のデータセットに対してfilterを行う処理の実行計画を比較する検証を実施しました。本記事では、その手順と実行結果を記述します。

optimizedPlanの実行計画ステージ以降で、Parquet形式ではサイズが変わらないのに対して、Delta Lake形式ではサイズが縮小することが判明しました。

| データフォーマット | 実行計画のデータサイズ |
| ------------------ | ----------------------: |
| Parquet形式        | 1,465,937,133          |
| Delta Lake形式     | 122,624                |

Delta Lake形式の実行計画のデータサイズ
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/57462328-ff04-1824-21e5-5d7a188febcd.png)

Parquet形式の実行計画のデータサイズ
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0bc0da58-245f-0edc-5fec-7ac3d4357058.png)


実行計画を比較すると、Delta Lake形式では`PreparedDeltaFileIndex`が利用されるなど実行計画が異なりました。

Delta Lake形式の実行計画
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/edff0df9-093a-172f-c05e-d10fcc184000.png)

Parquet形式の実行計画
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/dbcdd9a7-e35d-60cd-87e2-4de9988efff5.png)

詳細は下記のGithub pagesのページをご確認ください。

https://manabian-.github.io/databricks_tecks_for_qiita/#filtered_dataframe_size/filtered_dataframe_size.html

コードを実行したい方は、下記のdbcファイルを取り込んでください。

```URL
https://github.com/manabian-/databricks_tecks_for_qiita/blob/main/tecks/filtered_dataframe_size/dbc/filtered_dataframe_size.dbc
```

## 実行環境
databricks runtime: 9.0.x-scala2.12
Python  version   : 3.8.10
pyspark version   : 3.1.2

## 検証手順
### 事前準備

```python
filepath = "dbfs:/databricks-datasets/tpch/data-001/lineitem/"

schema = """
  L_ORDERKEY    INTEGER ,
  L_PARTKEY     INTEGER ,
  L_SUPPKEY     INTEGER ,
  L_LINENUMBER  INTEGER ,
  L_QUANTITY    DECIMAL(15,2) ,
  L_EXTENDEDPRICE  DECIMAL(15,2) ,
  L_DISCOUNT    DECIMAL(15,2) ,
  L_TAX         DECIMAL(15,2) ,
  L_RETURNFLAG  STRING ,
  L_LINESTATUS  STRING ,
  L_SHIPDATE    DATE ,
  L_COMMITDATE  DATE ,
  L_RECEIPTDATE DATE ,
  L_SHIPINSTRUCT STRING ,
  L_SHIPMODE     STRING ,
  L_COMMENT      STRING
"""
   
df = (spark
       .read
       .format("csv")
       .schema(schema)
       .option("sep", "|")
       .load(filepath)
    )
```

```python
dbutils.fs.rm('dbfs:/FileStore/qiita/filtered_dataframe_size', True)

# パーティションを設定したdelta形式のデータセットを準備
df.write.mode('overwrite').format('delta').partitionBy('L_SHIPDATE').save('dbfs:/FileStore/qiita/filetered_dataframe_size/lineitem_delta_partitionby')

# パーティションを設定したparquet形式のデータセットを準備
df.write.mode('overwrite').format('parquet').partitionBy('L_SHIPDATE').save('dbfs:/FileStore/qiita/filetered_dataframe_size/lineitem_parquet_partitionby')
```

### 実行計画のステージごとのサイズを表示

```python
# 関数を定義
def print_sizeinbytes(df):
    """複数の実行計画から取得できるデータのサイズを表示
    """
    table_size = df._jdf.queryExecution().logical().stats().sizeInBytes()
    print(f"logical       : {table_size}")

    table_size = df._jdf.queryExecution().analyzed().stats().sizeInBytes()
    print(f"analyzed      : {table_size}")

    table_size = df._jdf.queryExecution().withCachedData().stats().sizeInBytes()
    print(f"withCachedData: {table_size}")
    
    table_size = df._jdf.queryExecution().optimizedPlan().stats().sizeInBytes()
    print(f"optimizedPlan : {table_size}")

    table_size = df._jdf.queryExecution().sparkPlan().stats().sizeInBytes()
    print(f"sparkPlan     : {table_size}")
    
    table_size = df._jdf.queryExecution().executedPlan().stats().sizeInBytes()
    print(f"executedPlan  : {table_size}")
``` 

```python
# パーティションを設定したdelta形式のデータセットにフィルター処理した際のデータサイズ
df_2 = spark.read.load('dbfs:/FileStore/qiita/filetered_dataframe_size/lineitem_delta_partitionby').filter('L_SHIPDATE = CAST("1992-01-02" AS date)')

print_sizeinbytes(df_2)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/879113e3-554e-31cc-0da3-d428d94e8449.png)


```python
df_5 = spark.read.format('parquet').load('dbfs:/FileStore/qiita/filetered_dataframe_size/lineitem_parquet_partitionby').filter('L_SHIPDATE = CAST("1992-01-02" AS date)')
 
print_sizeinbytes(df_5)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/456f8815-c3a9-47f0-94c9-1c8ae2b0ef2c.png)