Databrikcs(Spark)のPysparkにて単一ファイルとしてCSVファイルを書き込む方法

## 概要
Databrikcs(spark)のPysparkにて単一ファイルのCSVファイルとして書き込む方法を共有します。
ただし、大量データの場合には、ファイルの単一化を実施すべきではないです。

下記の3つの設定を行うことで、単一ファイルとなります。

1. `spark.sql.sources.commitProtocolClass`を`org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol`に設定
2. `mapreduce.fileoutputcommitter.marksuccessfuljobs`を`false`に設定
3. coalesce関数によりパーティション数を`1`に設定

デフォルトの設定では、データを含むファイルが分割され、'_SUCCESS'などのファイルなどの無駄なファイルが書き込まれてしまいます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/78087e6f-3026-ab55-b2b4-9175ba21ba23.png)

本記事で紹介する方法により、単一ファイルのCSVファイルとして書き込むことができます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a4aad7cd-0e37-fdeb-65c8-8497543ae47a.png)

コードはGithub pagesにて公開しております。

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

実際に試したい方は、下記のファイルをインポートしてください。

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

## 実施手順
### Sparkコンフィグとパーティションの設定変更により、単一ファイルとして書き込む方法
下記のようにSparkコンフィグを変更します。

```python
# '_started'と'_committed_'で始まるファイルを書き込まないように設定
spark.conf.set("spark.sql.sources.commitProtocolClass", "org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol")

# '_SUCCESS'で始まるファイルを書き込まないように設定
spark.conf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs","false")
```

書き込み前にデータフレームのパーティション数を、coalesce関数により1にします。

```python
write_path = output_path + "/01"

dbutils.fs.rm(write_path, True)

(df.coalesce(1) # 書き込み時のファイル数を1つに再パーティション
     .write
     .format("csv")
     .option("compression","gzip")
     .option("sep", sep)
     .mode("overwrite")
     .save(write_path)
)

display(dbutils.fs.ls(write_path))
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a4aad7cd-0e37-fdeb-65c8-8497543ae47a.png)

### 指定数の複数ファイルとして書き込む方法
上記の設定変更を実施したうえで、repartition関数を利用することで、指定数の複数ファイルとして書き込めます。

```python
write_path = output_path + "/02"

dbutils.fs.rm(write_path, True)

(df.repartition(12)
     .write
     .format("csv")
     .option("compression","gzip")
     .option("sep", sep)
     .mode("overwrite")
     .save(write_path)
)

display(dbutils.fs.ls(write_path))
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d02cf8f2-3f03-1488-132e-ae5cf2002192.png)


### ファイル名の変更方法
ファイル名を変更する方法がSpark側の設定にはないため、Pythonにて名称を変更を実施する必要があります。
下記のコードでは、`ファイル名_yyyymmddhhMMssmmmmmm_###.tbl.gz`というファイル名に変更しております。

```python
import glob
import os
import datetime

file_path_for_python = f'{write_path}/*'.replace("dbfs:","/dbfs")

file_base_name = "customer"
file_name_datetime_string = datetime.datetime.now().strftime("%Y%m%d%H%M%S%f")
file_extension = ".tbl.gz"

# ファイル一覧の作成
file_list = glob.glob(file_path_for_python)

# ファイル一覧の名称を変更
for i, file_path in enumerate(file_list, 1):
    file_dir = os.path.dirname(file_path)
    
    new_file_path =os.path.join(file_dir,
                           file_base_name + 
                           "_" +
                           file_name_datetime_string + 
                           "_" +
                           '{0:03d}'.format(i) +
                           file_extension)
    os.rename(file_path, new_file_path)
    
    
display(dbutils.fs.ls(write_path))
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f2072a49-9ea5-764d-414f-faf03e29cff2.png)

## 参考情報
- [How do I prevent _success and _committed files in my write output? - Databricks Community Forum](https://forums.databricks.com/questions/14712/how-do-i-prevent-success-and-committed-files-in-my.html)