#3 Azure Databricks を使ってみよう 【格納編】


## 【 Azure Databricks ETL編 その３ 】

### 概要
Azure Databricks を使ってみよう！　ということで、以下の３ステップで簡単に說明します。
  1. [【環境準備編】](https://qiita.com/turupon/items/5a5782a62c24ccb3b3b5)：Azure CLI から Azure Databricks 環境を作成します
  2. [【抽出変換編】](https://qiita.com/turupon/items/21e2ca80ca497886d99d)：Databricks から ADLS Gen2 のデータを抽出し、変換します
  3. [【格納編】](https://qiita.com/turupon/items/50f3e12222fa2f7d4144)：Databricks で ETL されたデータを ADLS Gen2 に格納します

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/e6844375-e5e6-f5e7-0846-2fa60500e6b1.png)

今回は 【格納編】の說明となります。 Databricks の Notebook を使用していきます。

  - Azure Databricks
    - NoteBook 実装
      - Blob コンテナ（データ格納用）のマウント
      - データ格納

### ローカル環境

- macOS Big Sur 11.3 
- python 3.8.3
- Azure CLI 2.28.0

### 前提条件

1. Azure環境がすでに用意されていること（テナント/サブスクリプション）。
2. ローカル環境に「azure cli」がインストールされていること。
3. 本記事シリーズの[【環境準備編】](https://qiita.com/turupon/items/5a5782a62c24ccb3b3b5) と [【抽出変換編】](https://qiita.com/turupon/items/21e2ca80ca497886d99d) が完了していること

---

## Azure Databricks の Notebook を使います

[【抽出変換編】](https://qiita.com/turupon/private/e5ab639f7ad1980cbed2) で作成した Notebook :「ETL_UsageCost_01」 の続きとして実装します
なお、Blob コンテナのマウントについては、[この記事](https://qiita.com/KcMichael/items/8f5c8a3f97df31769f11) の内容をほぼそのまま活用させていただきました

### Blobコンテナーのマウント

```python:cmd_10
# コンテナーのマウント
# 一度マウントすると、Clusterを停止、変更してもマウント状態が維持されます
# マウントされた状態で再度操作を実行するとエラーが発生するため、マウント状態をチェックする

# Blob Storage情報（データ抽出用）
storage = {
  "account": "testaccount",
  "container": "etl-cost-data",
  "key": "xxxxxxxxxxxxxxxxxxxxxxxxxxxx"
}

# マウント先DBFSディレクトリ（データ抽出用）
mount_point_etl = "/mnt/etl-blob-data"

try:
  # マウント状態のチェック
  mount_dir = mount_point_etl
  if mount_dir[-1] == "/":
    mount_dir = mount_dir[:-1]
  if len(list(filter(lambda x: x.mountPoint == mount_dir, dbutils.fs.mounts()))) > 0:
    print("Already mounted.")
    mounted = True
  else:
    mounted = False

  # Blob Storageのマウント
  if not mounted:
    source = "wasbs://{container}@{account}.blob.core.windows.net".format(**storage)
    conf_key = "fs.azure.account.key.{account}.blob.core.windows.net".format(**storage)

    mounted = dbutils.fs.mount(
      source=source,
      mount_point = mount_point_etl,
      extra_configs = {conf_key: storage["key"]}
    ) 

except Exception as e:
  raise e

"mounted: {}".format(mounted)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/4116f73e-d0a4-0754-99ba-87ed424faea6.png)


### Blob Storageをマウントしたディレクトリの確認

```python:cmd_11
# ディレクトリの確認
display(dbutils.fs.mounts())
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/2ff942e9-3312-ea7d-b6c3-216b366703be.png)


### 変換されたデータの格納 : Json

```python:cmd_12
# データの保存：階層化指定なし
df.write.mode('overwrite').json("/mnt/etl-blob-data/NoPartition_01.json")

# データの保存：Subscription で階層化
df.write.mode('overwrite').partitionBy('Subscription').json("/mnt/etl-blob-data/ParSubscription_02.json")

# データの保存：Subscription - ResourceGroup で階層化
df.write.mode('overwrite').partitionBy('Subscription', 'ResourceGroup').json("/mnt/etl-blob-data/ParResourceGroup_03.json")
```

### 格納されたファイルの確認 : Json

```python:cmd_13
# 格納ファイルの確認（ファイルが存在しないとエラーが発生）
# mount_point = "/mnt/{マウント先ディレクトリ}"
display(dbutils.fs.ls(mount_point_etl))
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/f4ce1403-1adc-4c6f-fa88-1e61cb62e8ed.png)


```python:cmd_14
# 「階層化指定なし」で格納されたファイルの確認
display(dbutils.fs.ls("/mnt/etl-blob-data/NoPartition_01.json/"))
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/9a94e3f5-bf31-cb67-5245-12f1cc3d5235.png)


```python:cmd_15
# 「Subscription で階層化」で格納されたファイルの確認
display(dbutils.fs.ls("/mnt/etl-blob-data/ParSubscription_02.json"))
display(dbutils.fs.ls("/mnt/etl-blob-data/ParSubscription_02.json/Subscription=NSG-01"))
display(dbutils.fs.ls("/mnt/etl-blob-data/ParSubscription_02.json/Subscription=iapp-01"))
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/10e588e5-761b-8a22-a09c-8e0464c5e5c0.png)


```python:cmd_16
# 「Subscription - ResourceGroup で階層化」で格納されたディレクトリの確認
display(dbutils.fs.ls("/mnt/etl-blob-data/ParResourceGroup_03.json"))
display(dbutils.fs.ls("/mnt/etl-blob-data/ParResourceGroup_03.json/Subscription=NSG-01/"))
display(dbutils.fs.ls("/mnt/etl-blob-data/ParResourceGroup_03.json/Subscription=iapp-01/"))
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/0109952c-4b3b-6c66-06e6-c8ddbec82f78.png)


### 格納したファイルの再読み込み（確認） : Json

```python:cmd_17
# 「Subscription で階層化」で格納された全件データ読み込み
rdf = spark.read.json('/mnt/etl-blob-data/ParSubscription_02.json')
display(rdf)

# 「Subscription で階層化」で格納された「Subscription=iapp-01」のデータ読み込み
rdf = spark.read.json('/mnt/etl-blob-data/ParSubscription_02.json/Subscription=NSG-01')
display(rdf)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/d4e4d8d0-393b-4137-79ee-182b4e441219.png)



### 変換されたデータの格納 : Parquet

```python:cmd_18
# データの保存：階層化指定なし
df.write.mode('overwrite').parquet("/mnt/etl-blob-data/Parquet/NoPartition_11")

# データの保存：Subscription で階層化
df.write.mode('overwrite').partitionBy('Subscription').parquet("/mnt/etl-blob-data/Parquet/ParSubscription_12")
```

### 格納されたファイルの確認 : Parquet

```python:cmd_19
# 「Subscription で階層化」で格納されたファイルの確認
display(dbutils.fs.ls("/mnt/etl-blob-data/Parquet/ParSubscription_12"))
display(dbutils.fs.ls("/mnt/etl-blob-data/Parquet/ParSubscription_12/Subscription=NSG-01"))
display(dbutils.fs.ls("/mnt/etl-blob-data/Parquet/ParSubscription_12/Subscription=iapp-01"))
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/8949e4de-1203-150d-02d4-e74cdd17c564.png)


### 格納したファイルの再読み込み（確認） : Parquet

```python:cmd_20
# 「Subscription で階層化」で格納された全件データ読み込み
rdf = spark.read.parquet('/mnt/etl-blob-data/Parquet/NoPartition_11')
display(rdf)

# 「Subscription で階層化」で格納された「Subscription=iapp-01」のデータ読み込み
rdf = spark.read.parquet('/mnt/etl-blob-data/Parquet/ParSubscription_12/Subscription=NSG-01')
display(rdf)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/32e85483-537c-3f48-d922-858a33d9d29f.png)


## ローカル環境からコンテナー内の BLOB を一覧表示する

```
$ az storage blob list --account-name $STORAGE_ACCOUNT --container-name $ETL_CONTAINER --output table
```

---

### まとめ

Azure Databricks から Blob のデータを抽出し、そのデータを変換後、Blob に指定したパーティショニングで格納できることを確認できました

### 参考記事

以下の記事を参考にさせていただきました。感謝申し上げます
[Azure Databricks: 3-1. DBFSにBlob Storageをマウント](https://qiita.com/KcMichael/items/8f5c8a3f97df31769f11)