Delta Lakeの新機能：カラム名指定による統計情報の取得

## 概要

Databricks(Delta Lake)では、従来では統計情報を取得する際にはカラム位置を考慮する必要がありましたが、Databricks Runtime 13.2 （Delta Lake 3.0.0）移行では`delta.dataSkippingStatsColumns`プロパティによりカラム名を指定できるようになりました。

データスキップは、クエリのパフォーマンスを向上させるために不可欠なプロセスです。このプロセスでは、関連性のないデータの読み込みを避けることで、クエリの実行時間を短縮します。統計情報は、このデータスキップの効率を高めるために重要な役割を果たします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/70e24530-7ca9-cdf7-1b69-e3fc257f6847.png)

引用元：[Delta Lake に対するデータのスキップ - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/delta/data-skipping)

デフォルトでは前から 32 列の統計情報を取得するような仕様です。統計情報の取得対象のカラムを最適化する際には、ALTER 文によりカラム順を変更してから下記プロパティの値を設定する必要がありました。監査列をクエリで利用することがよくあり、本来は最後に配置したいにもかかわらず、統計情報を取得するために前の方に移動する必要がありました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7b706b58-d906-b40d-a8ea-39a02c1930f8.png)

引用元：[Delta テーブルのプロパティ リファレンス - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/delta/table-properties#delta-table-properties)

`delta.dataSkippingStatsColumns`プロパティによりカラム名を指定できるようになったため、カラム順を変更する必要がなくなりました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/143a54d3-0fa4-fccf-d2f0-b444f5829026.png)

引用元：[Delta Lake に対するデータのスキップ - Azure Databricks | Microsoft Learn](https://learn.microsoft.com/ja-jp/azure/databricks/delta/data-skipping#--specify-delta-statistics-columns)

統計情報に関する情報は、テーブルのディレクトリ配下`_delta_log`フォルダにある`crc`ファイルにて格納されます。`crc`ファイルは json 形式でデータを保持しており、parquet ファイルごとの統計情報を`stats`列にて保持しています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/36796a4f-1ef7-b391-e6dd-62f1982f201c.png)

本記事では、`delta.dataSkippingNumIndexedCols`プロパティと`delta.dataSkippingStatsColumns`プロパティの検証に利用したコードとその結果を共有します。

## 検証コードと実行結果

### 事前準備

スキーマを作成します。

```python
%sql
CREATE SCHEMA IF NOT EXISTS hive_metastore.stats_columns_test;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/24e5ec83-3bee-098b-2030-6965ad139579.png)

### `delta.dataSkippingNumIndexedCols`プロパティにてカラム名指定による統計情報の取得

テーブルを作成します。

```python
%sql
DROP TABLE IF EXISTS hive_metastore.stats_columns_test.dataSkippingStatsColumns_table_01;
CREATE TABLE hive_metastore.stats_columns_test.dataSkippingStatsColumns_table_01 (
  int_col int,
  str_col STRING,
  date_col date
)
TBLPROPERTIES (
  'delta.dataSkippingStatsColumns' = 'int_col,date_col'
)
;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5e170e21-e4c4-c03a-9f77-990b0a684187.png)

データを挿入します。

```python
%sql
INSERT INTO hive_metastore.stats_columns_test.dataSkippingStatsColumns_table_01
SELECT 1,'a',CAST('2020-01-01' AS date)
UNION ALL
SELECT 2,'b',CAST('2020-02-01' AS date)
UNION ALL
SELECT 3,'c',CAST('2020-03-01' AS date)
;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2be49e0c-5566-e9c2-3d19-ffbe6fa18886.png)

テーブルのディレクトリ（Loacation）を取得します。

```python
tgt_table_name = "hive_metastore.stats_columns_test.dataSkippingStatsColumns_table_01"

# テーブルの Location を取得
result = spark.sql(f"DESCRIBE EXTENDED {tgt_table_name}").collect()
table_location = next(row.data_type for row in result if row.col_name == "Location")
print(table_location)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2d918b37-d075-cb29-9a52-033c56fd8304.png)

テーブルのディレクトリ（Loacation）配下にある`_delta_log`フォルダのオブジェクトを表示します。crc ファイルの最新が`00000000000000000001.crc`であることを確認できます。

```python
fs_list = dbutils.fs.ls(f"{table_location}/_delta_log/")
display(fs_list)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/030ee9aa-e4ba-0836-6438-45523d273770.png)

`00000000000000000001.crc`のデータを表示します。

```python
import pprint
import json

file_name = "00000000000000000001.crc"
pprint.pprint(json.loads(dbutils.fs.head(f"{table_location}/_delta_log/{file_name}")))
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a8853c58-54a0-6490-0151-2924770e1c8d.png)

### `delta.dataSkippingStatsColumns`プロパティにてカラム名指定による統計情報の取得

テーブルを作成します。

```python
%sql
DROP TABLE IF EXISTS hive_metastore.stats_columns_test.dataSkippingNumIndexedCols_table_01;
CREATE TABLE hive_metastore.stats_columns_test.dataSkippingNumIndexedCols_table_01 (
  int_col int,
  str_col STRING,
  date_col date
)
TBLPROPERTIES (
  'delta.dataSkippingNumIndexedCols' = 2
)
;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3a6c5c25-c154-d0c6-a75c-a4a74cc82c16.png)

カラム順を変更します。

```python
%sql
ALTER TABLE hive_metastore.stats_columns_test.dataSkippingNumIndexedCols_table_01
CHANGE COLUMN date_col date_col date AFTER int_col;

-- カラム順が変更されていることを確認
DESC hive_metastore.stats_columns_test.dataSkippingNumIndexedCols_table_01;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b8e8410f-1d82-201a-2935-5fd593f09d00.png)

データを挿入します。

```python
%sql
INSERT INTO hive_metastore.stats_columns_test.dataSkippingNumIndexedCols_table_01 (int_col,str_col,date_col)
SELECT 1,'a',CAST('2020-01-01' AS date)
UNION ALL
SELECT 2,'b',CAST('2020-02-01' AS date)
UNION ALL
SELECT 3,'c',CAST('2020-03-01' AS date)
;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/41c078bf-9894-3e3c-4fb2-032559693f4c.png)

テーブルのディレクトリ（Loacation）を取得します。

```python
tgt_table_name = "hive_metastore.stats_columns_test.dataSkippingNumIndexedCols_table_01"

# テーブルの Location を取得
result = spark.sql(f"DESCRIBE EXTENDED {tgt_table_name}").collect()
table_location = next(row.data_type for row in result if row.col_name == "Location")
print(table_location)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c88b617e-9b07-a421-2bea-c970e8f0c747.png)

テーブルのディレクトリ（Loacation）配下にある`_delta_log`フォルダのオブジェクトを表示します。crc ファイルの最新が`00000000000000000002.crc`であることを確認できます。

```python
fs_list = dbutils.fs.ls(f"{table_location}/_delta_log/")
display(fs_list)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fc8256da-16c0-7198-d080-447b141d0afb.png)

`00000000000000000002.crc`のデータを表示します。

```python
import pprint
import json

file_name = "00000000000000000002.crc"
pprint.pprint(json.loads(dbutils.fs.head(f"{table_location}/_delta_log/{file_name}")))
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6f459e56-2486-719f-9735-ca30464c484e.png)

### その他の検証

#### `delta.dataSkippingStatsColumns`プロパティにて日本語カラムの指定可否

日本語のカラム名でも、想定通りに統計情報を取得できていること確認できました。

```python
%sql
DROP TABLE IF EXISTS hive_metastore.stats_columns_test.japanese_column;
CREATE TABLE hive_metastore.stats_columns_test.japanese_column (
  int_col int,
  str_col STRING,
  `日付列` date
)
TBLPROPERTIES (
  'delta.dataSkippingStatsColumns' = 'int_col,`日付列`'
);
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/311ac93e-a6ba-0177-895c-0cf5b1c75b5c.png)

```python
%sql
INSERT INTO hive_metastore.stats_columns_test.japanese_column (int_col,`日付列`,str_col)
SELECT 1,CAST('2020-01-01' AS date),'a'
UNION ALL
SELECT 2,CAST('2020-02-01' AS date),'b'
UNION ALL
SELECT 3,CAST('2020-03-01' AS date),'c' 
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d1007b23-a066-c937-f76b-dc7c9598e681.png)

```python
tgt_table_name = "hive_metastore.stats_columns_test.japanese_column"

import pprint
import json

# テーブルの Location を取得
result = spark.sql(f"DESCRIBE EXTENDED {tgt_table_name}").collect()
table_location = next(row.data_type for row in result if row.col_name == "Location")
print(table_location)

file_name = "00000000000000000001.crc"
pprint.pprint(json.loads(dbutils.fs.head(f"{table_location}/_delta_log/{file_name}")))
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/768ea9bd-7979-5c42-160f-261dda0af9e8.png)

#### crc ファイルをデータフレームで表示

下記のコード群にて、crc ファイルをデータフレームとして表示します。

```python
tgt_table_name = "hive_metastore.stats_columns_test.dataSkippingStatsColumns_table_01"

# テーブルの Location を取得
result = spark.sql(f"DESCRIBE EXTENDED {tgt_table_name}").collect()
table_location = next(row.data_type for row in result if row.col_name == "Location")
print(table_location)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a7430763-2601-998e-10b9-d26221c7b14d.png)

```python
from pyspark.sql.functions import expr

df = spark.read.json(f"{table_location}/_delta_log/*.crc")

# 追加するカラムの定義
cols_conf = {
    "file_name": expr("_metadata.file_name"),
}

# カラムの並び順を考慮したリストを定義
df_columns = list(cols_conf.keys())
df_columns.extend(df.columns)

df = df.withColumns(cols_conf)
df = df.select(df_columns)

df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/cd5d4b0a-ddd1-d6ea-00a9-e6ba3e910cc2.png)

```python
path_df = spark.sql(
    """
    WITH src AS (
    SELECT
        file_name,
        explode(allFiles) AS allFiles
    FROM
        {df}
    )
    SELECT
        file_name
        ,collect_list(struct(allFiles.path, allFiles.stats)) as paths_and_stats
    FROM
        src
    GROUP BY
        ALL
    """,
    df=df,
)

path_df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/be4da169-b0ab-20cc-def8-c749a3180df4.png)

### 事後処理

スキーマを削除します。

```python
%sql
DROP SCHEMA hive_metastore.stats_columns_test CASCADE;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fb6211a5-b603-245c-ad4e-9ec85ea74554.png)