Sparkのファイルとパーティションの関係について確認してみた

# 背景・目的
Sparkでファイルを読み込んだ直後に、RDDで展開されるパーティション数について気になったので整理してみました。

# まとめ
- Sparkでファイルを読み込んだ直後のパーティション数は、ファイル数とファイルフォーマット、ファイルサイズとSparkパラメータの組み合わせで決まります。
    - ファイル数
        - 基本的に1ファイル
    - 分割可能なファイルフォーマットである。下記のようなフォーマットが対象
        - JSON
        - Parquet
        - ORC
    - ファイルサイズ > パラメータ(spark.sql.files.maxPartitionBytes)のときに、ファイル分割されます。
- 以下のような検証を行い、想定通りの結果になりました。
    | シナリオ | ファイル数 | ファイルフォーマット | ファイルサイズ|Sparkパラメータ<BR>`spark.sql.files.maxPartitionBytes` | **結果<BR>（パーティション数）**|
    |:--- |:--- |:--- |:--- |:--- |:--- |
    | 128MB以下の場合 | 1 | CSV | 約1MB | 128MB | 1 |
    | 1ファイル1パーティションである | 2 | CSV | 約1MB | 128MB | 2 |
    | 分割されないフォーマット| 1 | CSV | 約1MB | 1KB   | 1 |
    | 分割されるフォーマット | 1 | Parquet | 約1MB | 1KB | 840 |
    | パラメータの有効性の確認 | 1 | Parquet | 約1MB | 2KB | 420 |


# 概要
## ファイル分割について
[Other Configuration Options](https://spark.apache.org/docs/latest/sql-performance-tuning.html#other-configuration-options)には、下記の記述があります。

### spark.sql.files.maxPartitionBytes	
> ファイルの読み取り時に 1 つのパーティションにパックする最大バイト数。 この構成は、Parquet、JSON、ORC などのファイルベースのソースを使用する場合にのみ有効です。
- パラメータが有効なファイルフォーマットは下記の通り
    - Parquet
    - JSON
    - ORC
- デフォルトは、134217728 (128 MB)


# 実装
## 事前準備
- S3に、下記のファイルを配置し検証します。
    - parquetとtextファイル（CSV）を配置するディレクトリを2つ用意します。
        ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/9af2c2e9-5730-4b69-1eb6-82befd1eb63c.png)
    - textディレクトリ（CSVファイルを1つのみ配置）
        ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/4462d3d4-36ee-bbf2-4713-b0895b8f4dbc.png)
    - text-2ディレクトリ（CSVファイルを2つ配置）
        ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/76b14f28-0c1c-4825-ac05-e962a2a3de7e.png)
    - Paruetディレクトリ（Parquetファイルを1つ配置）
        ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/84b375bd-282a-df3e-55c3-0f89907b6a37.png)
- Glue Jupyter Notebookで試しています。


## シナリオ
下記の検証を行い、パーティション数の確認を行います。

1. 1MBのCSVファイルを読み込んだときのパーティション数→1パーティションになる想定
1. 1MBのCSVファイル２つを読み込んだときのパーティション数→2パーティションになる想定
1. spark.sql.files.maxPartitionBytesを1024に設定& 1MBのCSVファイルを読み込んだときのパーティション数→1パーティションになる想定
1. spark.sql.files.maxPartitionBytesを1024に設定& 1MB弱（840KB）のParquetファイルを読み込んだときのパーティション数→複数パーティションになる想定
1. spark.sql.files.maxPartitionBytesを2048に設定& 1MB弱（840KB）のParquetファイルを読み込んだときのパーティション数→上記4の半分になる想定


###  1. 1MBのCSVファイルを読み込んだときのパーティション数→1パーティションになる想定
:::note info
ファイルが1つのため、想定通り1パーティション
:::
```
# 1MBファイルを一つ
from pyspark.sql import types as T, functions as F

print(spark.conf.get('spark.sql.files.maxPartitionBytes'))

S3bucket_node1 = glueContext.create_dynamic_frame.from_options(
    format_options={"multiline": False},
    connection_type="s3",
    format="csv",
    connection_options={"paths": ["s3://XXXX/text/"], "recurse": True},
    transformation_ctx="S3bucket_node1",
)

S3bucket_node1.printSchema()
S3bucket_node1.toDF().rdd.getNumPartitions()

===

1024
root
|-- col0: string

1
```

### 2.1MBのCSVファイル２つを読み込んだときのパーティション数→2パーティションになる想定
::: note info
ファイルが2つのため、想定通り2パーティション
:::
```
# 1MBファイルを２つ→2パーティション
from pyspark.sql import types as T, functions as F

print(spark.conf.get('spark.sql.files.maxPartitionBytes'))


S3bucket_node2 = glueContext.create_dynamic_frame.from_options(
    format_options={"multiline": False},
    connection_type="s3",
    format="csv",
    connection_options={"paths": ["s3://XXXX/text-2/"], "recurse": True},
    transformation_ctx="S3bucket_node2",
)

S3bucket_node2.printSchema()
S3bucket_node2.toDF().rdd.getNumPartitions()

===

1024
root
|-- col0: string

2
```

### 3. spark.sql.files.maxPartitionBytesを1024に設定& 1MBのCSVファイルを読み込んだときのパーティション数→1パーティションになる想定
::: note info
分割不可能なファイルのため、想定通り1パーティション
:::

```
# maxPartitionBytesを設定する
from pyspark.sql import types as T, functions as F

# 1024 Byte単位にファイル分割できるか？
spark.conf.set('spark.sql.files.maxPartitionBytes',1024)
print("変更後：{}".format(spark.conf.get('spark.sql.files.maxPartitionBytes')))

S3bucket_node3 = glueContext.create_dynamic_frame.from_options(
    format_options={"multiline": False},
    connection_type="s3",
    format="csv",
    connection_options={"paths": ["s3://XXXX/text/"], "recurse": True},
    transformation_ctx="S3bucket_node3",
)

S3bucket_node3.printSchema()
S3bucket_node3.toDF().rdd.getNumPartitions()

===
変更後：1024
root
|-- col0: string

1

```

### 4. spark.sql.files.maxPartitionBytesを1024に設定& 1MB弱（840KB）のParquetファイルを読み込んだときのパーティション数→複数パーティションになる想定
:::note info
分割可能なファイルフォーマットで、ファイルサイズ >spark.sql.files.maxPartitionBytesが成り立つため、パーティション数は分割される。 
:::
```
# maxPartitionBytesを設定する
from pyspark.sql import types as T, functions as F

# 1024 Byte単位にファイル分割できるか？→840パーティション
spark.conf.set('spark.sql.files.maxPartitionBytes',1024)
print("変更後：{}".format(spark.conf.get('spark.sql.files.maxPartitionBytes')))

S3bucket_node4 = glueContext.create_dynamic_frame.from_options(
    format_options={"multiline": False},
    connection_type="s3",
    format="parquet",
    connection_options={"paths": ["s3://XXXX/parquet/"], "recurse": True},
    transformation_ctx="S3bucket_node4",
)

S3bucket_node4.printSchema()
S3bucket_node4.toDF().rdd.getNumPartitions()

===
変更後：1024
root
|-- col0: string

840

```

### 5. spark.sql.files.maxPartitionBytesを2048に設定& 1MB弱（840KB）のParquetファイルを読み込んだときのパーティション数→上記4の半分になる想定
:::note info
4と比較して、パラメータの値を倍にしたため、パーティション数は想定通り半分になる。
:::

```
# maxPartitionBytesを設定する
from pyspark.sql import types as T, functions as F

# 2048 Byte単位にファイル分割できるか？
spark.conf.set('spark.sql.files.maxPartitionBytes',2048)
print("変更後：{}".format(spark.conf.get('spark.sql.files.maxPartitionBytes')))

S3bucket_node5 = glueContext.create_dynamic_frame.from_options(
    format_options={"multiline": False},
    connection_type="s3",
    format="parquet",
    connection_options={"paths": ["s3://XXXX/parquet/"], "recurse": True},
    transformation_ctx="S3bucket_node5",
)

S3bucket_node5.printSchema()
S3bucket_node5.toDF().rdd.getNumPartitions()

===

変更後：2048
root
|-- col0: string

420

```

# 考察
ファイルフォーマットとファイルサイズにより、Sparkのパーティション数が変わることがわかりました。

# 参考
https://qiita.com/zumax/items/b527509047f219fcee39

https://spark.apache.org/docs/latest/sql-performance-tuning.html