Azure Databricks で単一のJSONファイルにデータを書き込んでみました

## 概要

Azure Databricks のデータ書き込みにおいて、シングルファイルで可能かどうかという依頼を頂いたので、サクッと検証した結果報告となります。 Spark と Pandas で少し結果が異なるようです。

### ローカル環境

- macOS Monterey 12.1 
- python 3.8.12

### 前提条件

1. Azure Databricks の環境が準備できていること - [参考](https://qiita.com/turupon/items/302666842f97206ff0f2)
2. Azure CostManagement 等のデータがBLOBに蓄積されていること（JSONであれば問わず） - [参考](https://qiita.com/turupon/items/a0cc580e26980ecc5c81)
3. Azure Databricks から 上記 BLOB のコンテナをマウントできていること
   - 読み込みコンテナのマウントポイント：/mnt/blob-data
   - 書き込みコンテナのマウントポイント：/mnt/blob-data2

---

## データの準備

### Blob Storageをマウントした ディレクトリ と ファイル の確認

```python:cmd_1
# マウント先DBFSディレクトリ（データ読み込み用）
mount_point = "/mnt/blob-data"

# マウント先DBFSディレクトリ（データ書き込み用）
mount_point2 = "/mnt/blob-data2"

# ディレクトリの確認
display(dbutils.fs.mounts())

# ファイルの確認（ファイルが存在しないとエラーが発生）
# mount_point = "/mnt/{マウント先ディレクトリ}"
display(dbutils.fs.ls(mount_point))
display(dbutils.fs.ls(mount_point2))
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/cf2e787e-362a-1116-6f81-1d64dfc27de9.png)


### データの読み込み（＋マージ）

```python:cmd_2
# Blob から 全ファイルの読込 --> PySpark Dataframes 型式
# （読み込むファイルの指定をワイルドカードにすると、複数ファイルを同時に扱えます）
storename = mount_point + '/*.json'
sdf = spark.read\
    .option("multiline", "true").option("header", "false").option("inferSchema", "false")\
    .json(storename)

display(sdf)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/f5e51d36-0962-b5fa-bc65-aa268ec4b7c2.png)


### 読み込んだデータの変換

```python:cmd_3
import pandas as pd
import datetime

# PySpark Dataframes から Pandas への変換
pdf = sdf.toPandas()

# 日付(date)をint64型からdatetime64型に変換
print(pdf['Date'].dtype)  # 変換前のデータ型
pdf['Date'] = pd.to_datetime(pdf['Date'].astype(str), format='%Y-%m-%d')
print(pdf['Date'].dtype)  # 変換後のデータ型

# Pandas から PySpark Dataframes への変換
df = spark.createDataFrame(pdf)
display(df)
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/9fdf428a-6794-6d7b-a374-6baf0c29d47a.png)


---

## データの書き込み

### Spark to Write Data to Multi JSON files （データの容量により複数ファイルに分割される）

```python:cmd_4
# ファイル名の作成
storename = mount_point2 + '/MargeUsageCost_' + datetime.datetime.now().strftime('%Y%m%d_%H%M%S') + '.json'
print(storename)

# データ保存：複数ファイルの分割されます
df.write.mode('overwrite').json(storename)

# データ確認
display(dbutils.fs.ls(mount_point2))
display(dbutils.fs.ls(storename))
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/68a3508a-0224-63c9-d74c-56bc7f40fc44.png)


### Spark to Write Data  to a Single JSON file （Use the coalesce Feature）

```python:cmd_5
# ファイル名の作成
storename = mount_point2 + '/MargeUsageCost_' + datetime.datetime.now().strftime('%Y%m%d_%H%M%S') + '.json'
print(storename)

# データ保存：シングルファイルですが、ディレクトリが作成されます
df.coalesce(1).write.mode('overwrite').json(storename)

# データ確認
display(dbutils.fs.ls(mount_point2))
display(dbutils.fs.ls(storename))
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/a7e24edb-dd2a-b7bd-2ca9-1a155f6ed063.png)


### Pandas to Write Data to a Single JSON file （Use collect and Pandas）

```python:cmd_6
# ファイル名の作成
storename = '/dbfs' + mount_point2 + '/MargeUsageCost_' + datetime.datetime.now().strftime('%Y%m%d_%H%M%S') + '.json'
print(storename)

# PySpark Dataframes から Pandas への変換
dfp = df.toPandas()

# データ保存：マウントポイントにシングルファイルで書き込みされます
# orientを指定しない場合、orient='columns'を指定したときと同じ内容でjson出力されます
# dfp.to_json(storename, orient='columns', date_format='iso')  # 列で書き込む場合
dfp.to_json(storename, orient='records', date_format='iso')  # 行で書き込む場合

# データ確認
display(dbutils.fs.ls(mount_point2))
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/544022/3d2edf76-10e5-19f4-823d-2650d744298f.png)


---

### まとめ

Azure Databricks の書き込みにおいて、通常複数ファイルに分割されますが、シングルファイルでできる方法を確認できました。

### 参考記事

以下の記事を参考にさせていただきました。感謝申し上げます
[Write to a Single CSV File(Python)](https://analyticjeremy.github.io/Databricks_Examples/Write%20to%20a%20Single%20CSV%20File)
[pandas.DataFrame.to_json](https://pandas.pydata.org/pandas-docs/version/1.3.4/reference/api/pandas.DataFrame.to_json.html)