Spark のデータオブジェクト のメタデータ取得方法

## 概要

Spark のデータオブジェクトのメタデータを、PySpark 、および、Spark SQL により取得する方法を説明する。本記事では、Databricks での実行結果を掲載。

## 本記事の位置付け

次の開発ガイドシリーズにおけるメタデータデプロイ分野の補足記事であり、リンク先には記事にて記事の全体像を整理している。

| GroupID | 分野               |
| ------- | ---------------------- |
| T10     | Spark概要                  |
| T20     | データエンジニアリング |
| T30     | データ品質チェック     |
| T40     | データサイエンス       |                                         |
| **T50**     | **メタデータデプロイ**     |
| T60     | テスト                 |
| T70     | DevOps                 |

https://qiita.com/manabian/items/67bcb6c5823bc4d5d944

## 事前準備

```python
from pyspark.sql import Row
from decimal import *
import datetime
import pprint
```

```python
import random, string
random_string = ''.join(random.choices(string.ascii_letters, k=5))
```

```python
db_name = f'_metadata_test_database_{random_string}'
print(db_name)
```

```python
# データベースを作成
sql = f'DROP DATABASE IF EXISTS {db_name} CASCADE'
spark.sql(sql)

sql = f'CREATE DATABASE {db_name}'
spark.sql(sql)
```

```python
table_name      = '_metadata_test_table'
table_full_name = f'{db_name}.{table_name}'

sample_Data = [
    Row(
        string_column    = 'AAA', 
        byte_column      = 1, 
        integer_column   = 1, 
        bigint_column    = 1, 
        float_column     = 12.300000190734863, 
        double_column    = 12.3, 
        numeric_column   = Decimal('12'), 
        boolean_column   = True, 
        date_column      = datetime.date(2020, 1, 1), 
        timestamp_column = datetime.datetime(2021, 1, 1, 0, 0), 
        binary_column    = bytearray(b'A'), 
        struct_column    = Row(
            struct_string_column = 'AAA', 
            struct_int_column    = 1
        ), 
        array_column     = ['AAA', 'BBB', 'CCC'], 
        map_column       = {'AAA': 1}
    )
]


schema = '''
--文字型
string_column string,

--整数型
byte_column byte,
integer_column integer,
bigint_column bigint,

--浮動小数点型
float_column float,
double_column double,
numeric_column numeric,

--真偽型
boolean_column boolean,

--日付時刻
date_column date, 
timestamp_column timestamp,

--バイナリー型
binary_column binary,

--複合型
struct_column struct<
    struct_string_column :string,
    struct_int_column    :int
>,
array_column array<string>, 
map_column map<string, int>
'''

df = spark.createDataFrame(sample_Data, schema)
df.display()

# テーブルを作成
spark.sql(f'DROP TABLE IF EXISTS {table_full_name}')
df.write.format('delta').saveAsTable(table_full_name)
```

```python
table_name_02      = '_metadata_test_table_02'
table_full_name_02 = f'{db_name}.{table_name_02}'

sql = f'''
DROP TABLE IF EXISTS {table_full_name_02}
'''
spark.sql(sql)

sql = f'''
CREATE TABLE {table_full_name_02}
(
    id BIGINT GENERATED ALWAYS AS IDENTITY
    ,string_column string
    ,integer_column integer
    ,date_column date
    ,generated_col INT GENERATED ALWAYS AS (integer_column)
)
USING delta
TBLPROPERTIES (
    delta.autoOptimize.optimizeWrite = True, 
    delta.autoOptimize.autoCompact   = True,
    delta.dataSkippingNumIndexedCols = 1
  )
PARTITIONED BY (
    string_column
)
COMMENT "table_comment"
'''
spark.sql(sql)

```

```python
table_name_03      = '_metadata_test_table_03'
table_full_name_03 = f'{db_name}.{table_name_03}'

sql = f'''
DROP TABLE IF EXISTS {table_full_name_03}
'''
spark.sql(sql)

sql = f'''
CREATE OR REPLACE TABLE {table_full_name_03}
(
    col_001 string
    ,col_002 string
    ,col_003 string
)
USING delta
TBLPROPERTIES (
    delta.autoOptimize.optimizeWrite = True, 
    delta.autoOptimize.autoCompact   = True,
    delta.dataSkippingNumIndexedCols = 1
  )
'''
spark.sql(sql)

```

```python
view_name      = '_v_metadata_test_table'
view_full_name = f'{db_name}.{view_name}'

sql = f'''
CREATE OR REPLACE VIEW {view_full_name}
AS
SELECT
  *
  FROM
    {table_full_name}
'''

spark.sql(sql)
```


## Spark Schema (Database) のメタデータ取得方法

### DESCRIBE DATABASE

```python
sql = f"DESCRIBE DATABASE {db_name}"

spark.sql(sql).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a5fea935-77a6-5c7c-ba1f-17ff5da3eb57.png)


### SHOW SCHEMAS

```python
sql = f"SHOW SCHEMAS LIKE '{db_name}'"

spark.sql(sql).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d7249da3-9de2-8bdc-41e7-7530d0095967.png)

### SHOW TABLES

```python
sql = f"SHOW TABLES FROM {db_name}"

spark.sql(sql).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/cf250762-bd7f-9d8d-3baf-549be45b23ae.png)

### SHOW VIEWS

```python
sql = f"SHOW VIEWS FROM {db_name}"

spark.sql(sql).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0dfabf76-d37a-e892-d5c7-774be4ea3a9e.png)

## Spark Table のメタデータ

### SHOW CREATE TABLE

```python
sql = f'SHOW CREATE TABLE {table_full_name}'

spark.sql(sql).display()
print(spark.sql(sql).collect()[0][0]) 
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/00b937ab-bb16-3813-a725-0cd5f68e24ca.png)

### DESCRIBE

```python
sql = f'DESCRIBE {table_full_name}'

spark.sql(sql).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ba3a4d82-3b48-3a70-8006-1171607b7080.png)

```python
sql = f'DESCRIBE {table_full_name} integer_column'

spark.sql(sql).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2f6487fb-a073-0296-3623-431050be990f.png)

### DESCRIBE EXTENDED

```python
sql = f'DESCRIBE EXTENDED {table_full_name}'

spark.sql(sql).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b5675c3d-0e91-0e58-8d52-3443af4e5511.png)

```python
sql = f'DESCRIBE EXTENDED {table_full_name} integer_column'

spark.sql(sql).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/bd5cc043-090a-6af1-ed00-a8e4b75b5c37.png)

```python
# 統計情報取得後に、カラムの統計情報を取得
sql = f'ANALYZE TABLE {table_full_name} COMPUTE STATISTICS FOR COLUMNS integer_column'
spark.sql(sql).display()
sql = f'DESCRIBE EXTENDED {table_full_name} integer_column'
spark.sql(sql).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9b491234-cdb2-3ca7-08ef-38a0a0f15055.png)

### DESCRIBE QUERY

```python
sql = f'''
DESCRIBE QUERY
SELECT
  string_column
  ,integer_column
  ,COUNT(*) AS COUNT

  FROM 
    {table_full_name}
  GROUP BY
    string_column
    ,integer_column

'''

spark.sql(sql).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3f711401-ebb5-ca88-c91b-c147a37352ba.png)

### SHOW COLUMNS

```python
sql = f'SHOW COLUMNS IN {table_full_name}'

spark.sql(sql).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8dee341a-9f12-9c7e-abf4-79f9710f35bb.png)

### SHOW TBLPROPERTIES

```python
sql = f'SHOW TBLPROPERTIES {table_full_name}'

spark.sql(sql).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4c31d963-04d9-a88d-a252-719e41df30ee.png)

## Delta Lake テーブルのメタデータ取得

### DESCRIBE DETAIL

```python
sql = f'DESCRIBE DETAIL {table_full_name}'

spark.sql(sql).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9dbe089d-8ffe-7858-cd9b-cb851af3b79c.png)

### DESCRIBE HISTORY

```python
sql = f'DESCRIBE HISTORY {table_full_name}'

spark.sql(sql).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/cadd654e-2290-d3c4-7afe-441a79462a36.png)


### `Delta Transaction Log`上のメタデータ

```python
import json

# 最新の Delta Transaction Log の version を取得
max_version = (
    spark
    .sql(f'DESC HISTORY {table_full_name_02}')
    .sort('version', ascending=False)
    .select('version')
    .limit(1)
    .collect()[0][0]
)

# Delta Lake テーブルのファイルパスを取得
file_path = spark.sql(f'DESC EXTENDED {table_full_name_02}').where("col_name = 'Location'").select('data_type').collect()[0][0]

file_content = dbutils.fs.head(f'{file_path}/_delta_log/{max_version:020}.json')

# １つファイルに内に格納される json 文字列群から、`metaData`の文字列を取得
delta_log_metadata = json.loads(
    [l for l in file_content.split('\n') if l != '' and json.loads(l).get('metaData') != None][0]
)

pprint.pprint(delta_log_metadata)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/88e98f62-8eda-1c05-923a-90c6de41e1fd.png)

## Spark Dataframe のメタデータ

### DataFrame.printSchema

```python
df.printSchema()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/aa1816b7-b249-a49f-80e4-0952d908845b.png)

### DataFrame.schema

```python
df.schema
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/adbebf3c-d3c0-c59a-16a7-2d2e68fd08e9.png)

### toDDL(DDL 文字列)

```python
df_schema = df.schema
pprint.pprint(spark.sparkContext._jvm.org.apache.spark.sql.types.DataType.fromJson(df_schema.json()).toDDL())
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/807cd777-1662-6aa7-335f-8e9c540c0c2f.png)

### DataFrame.columns

```python
df.columns
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9cf44897-97d0-4d3b-39a3-50e6a9496017.png)

### DataFrame.dtypes

```python
df.dtypes
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/582d8d9e-804a-31e0-1e7a-42dc238dac71.png)

### DataFrame.storageLevel

```python
df.storageLevel
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5f1ceea7-75e9-bebb-cda8-d86a6f625997.png)

### DataFrame.inputFiles

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

schema = """
    r_regionkey date,
    r_name string,
    r_comment string
"""
   
df_2 = (
    spark
    .read
    .format("csv")
    .schema(schema)
    .option("sep", "|")
    .load(filepath)
)

df_2.inputFiles()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/83228a6f-ded3-8709-ebe1-2d352a9eceb0.png)

### Spark Config の設定値

```python
spark_confs = spark.sparkContext.getConf().getAll()

spark.createDataFrame(spark_confs).limit(5).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/06615e69-f5ed-1db9-c84b-1067570ebda4.png)

```python
sql = 'set -v'

spark.sql(sql).limit(5).display()
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/17c5fc95-a844-4a32-6fda-222037eb3043.png)

## Databricks 固有機能

### SHOW GRANTS ON

```python
# 下記については、ユーザー名（メールアドレス）が表示されるため実行結果共有時に注意が必要
try:
    sql = f'SHOW GRANTS ON {table_full_name}'
    spark.sql(sql).display()
except:
    print('Databricks の テーブルアクセスコントロールができるクラスターで実行してください。')
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b5639257-84f0-345a-39dc-b63e6f78485f.png)


## リソースのクリーンアップ

```python
# データベースを作成
sql = f'DROP DATABASE IF EXISTS {db_name} CASCADE'
spark.sql(sql)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0dead2a0-5b6a-6d62-d7ac-d895a1ec9011.png)