Databricks(Spark）のPythonにてブルームフィルターが設定されているカラム情報を取得する方法

## 概要
Databricks(Spark）のPythonにてブルームフィルターが設定されているカラム情報を取得する方法を共有します。

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

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

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

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

## 環境
databricks runtime: 8.3.x-cpu-ml-scala2.12
Python  version:    3.8.8
pyspark version:    3.1.2.dev0

## 手順
### 事前準備
検証に利用するデータベースとテーブルを作成します。

```python
dbutils.fs.rm("/Filestore/qiita/bloom", True)
```

```python
%sql
 
create database if not exists qiita;
 
drop table if exists qiita.bloomfilter_table;
create table qiita.bloomfilter_table
(
  string long
  ,number int
  ,date date
)
using delta
location "/Filestore/qiita/bloom"
;

CREATE BLOOMFILTER INDEX
ON TABLE qiita.bloomfilter_table
FOR COLUMNS(string OPTIONS (fpp=0.1, numItems=50000000))

CREATE BLOOMFILTER INDEX
ON TABLE qiita.bloomfilter_table
FOR COLUMNS(number OPTIONS (fpp=0.1, numItems=10000000))
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ef1d05f7-b6ac-8efa-8cff-fe8ad5a86034.png)

### 関数を定義

```python
def get_defination(schema, prefix=''):
    outstring_format='"name":{0} nullable:{1}'
    fields = schema['fields']
    list_definations=[]
    prefix = prefix+'.' if prefix !='' else ''
    for field in fields:
        if type(field['type']) is dict:
            if field['type']['type'] == 'struct':
                list_definations.extend(get_defination(field['type'], prefix+field['name']))
            else:
                field["name"]=prefix + field["name"]
                list_definations.append(field)
        else:
            field["name"]=prefix + field["name"]
            list_definations.append(field)
    return list_definations

def get_bloom_filter_columns_info(schema_json):
    allcols = get_defination(schema_json)
    return list(filter(lambda col: col['metadata'].get('delta.bloomFilter.enabled') is not None, allcols))
```

### 階層をフラットにしたDDLのリスト型変数を取得

```python
## dict形式のddl文を取得
table_name = 'qiita.bloomfilter_table'
schema_dict = spark.table(table_name).schema.jsonValue()

not_null_columns_list = get_bloom_filter_columns_info(schema_dict)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/44665761-2bb5-c04a-14aa-24e24fcf3550.png)

### 利用例）リスト型変数からDDL文を作成

```python
# 利用例
outstring_format='''
CREATE BLOOMFILTER INDEX
ON TABLE {0}
FOR COLUMNS({1} OPTIONS (fpp={2}, numItems={3}))
'''.strip()
 
for l in not_null_columns_list:
    name = l['name']
    sample_string = outstring_format.format(table_name,
                                            l['name'],
                                            l['metadata'].get('delta.bloomFilter.fpp'),
                                            l['metadata'].get('delta.bloomFilter.numItems'),
                                           )
    print('---ddl_sample---')
    print(sample_string)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/c46c3450-eebc-3fb1-988c-7835c463ab29.png)