SparkのパーティションがParquetファイルの保存に与える影響

こちらの続きです。ここで書いてあることは、Delta Lakeにも当てはまります。

https://qiita.com/taka_yayoi/items/2b30335926a9ac31f3b3

ノートブックは[こちら](https://github.com/datanikkthegreek/SparkDeltaDatabricksInternals/blob/main/Lection2%20-%20Spark%20Partitions/S02E05_HowPartitionsInfluenceSavingData.ipynb)で、動画はこちらです。

<iframe width="560" height="315" src="https://www.youtube.com/embed/zfEuMNh01Uk?si=awOVbBf-_s0lpDDJ" title="YouTube video player" frameborder="0" allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share" referrerpolicy="strict-origin-when-cross-origin" allowfullscreen></iframe>

# 0. セットアップ

このノートブックの一般的なヒント:
- Spark UIは**クラスター -> Spark UI**でアクセス可能
- Spark UIの詳細な調査は後のエピソードで行います
- `sc.setJobDescription("Description")`はSpark UIのアクションのジョブ説明を独自のものに置き換えます
- `sdf.rdd.getNumPartitions()`は現在のSpark DataFrameのパーティション数を返します
- `sdf.write.format("noop").mode("overwrite").save()`は、実際の書き込み中に副作用なしで変換を分析および開始するための良い方法です

```py
from pyspark.sql import SparkSession
from pyspark.sql import functions as f
```

```py
# AQEをオフにする。これは、このシナリオでは混乱を招く可能性があるため、追加のジョブを生成しないようにする。
spark.conf.set("spark.sql.adaptive.enabled", "false")
# データフレームをキャッシュしないようにする...これにより、繰り返し可能な結果が得られない場合があります。
spark.conf.set("spark.databricks.io.cache.enabled", "false")
```

```py
def sdf_generator(num_rows: int, num_partitions: int = None) -> "DataFrame":
    return (
        spark.range(num_rows, numPartitions=num_partitions)
        .withColumn("date", f.current_date())
        .withColumn("timestamp",f.current_timestamp())
        .withColumn("idstring", f.col("id").cast("string"))
        .withColumn("idfirst", f.col("idstring").substr(0,1))
        .withColumn("idlast", f.col("idstring").substr(-1,1))
        )
```

```py
sdf_gen = sdf_generator(20)
sdf_gen.count()
```
```
20
```

```py
sdf_gen.show()
```
```
+---+----------+--------------------+--------+-------+------+
| id|      date|           timestamp|idstring|idfirst|idlast|
+---+----------+--------------------+--------+-------+------+
|  0|2024-12-05|2024-12-05 04:19:...|       0|      0|     0|
|  1|2024-12-05|2024-12-05 04:19:...|       1|      1|     1|
|  2|2024-12-05|2024-12-05 04:19:...|       2|      2|     2|
|  3|2024-12-05|2024-12-05 04:19:...|       3|      3|     3|
|  4|2024-12-05|2024-12-05 04:19:...|       4|      4|     4|
|  5|2024-12-05|2024-12-05 04:19:...|       5|      5|     5|
|  6|2024-12-05|2024-12-05 04:19:...|       6|      6|     6|
|  7|2024-12-05|2024-12-05 04:19:...|       7|      7|     7|
|  8|2024-12-05|2024-12-05 04:19:...|       8|      8|     8|
|  9|2024-12-05|2024-12-05 04:19:...|       9|      9|     9|
| 10|2024-12-05|2024-12-05 04:19:...|      10|      1|     0|
| 11|2024-12-05|2024-12-05 04:19:...|      11|      1|     1|
| 12|2024-12-05|2024-12-05 04:19:...|      12|      1|     2|
| 13|2024-12-05|2024-12-05 04:19:...|      13|      1|     3|
| 14|2024-12-05|2024-12-05 04:19:...|      14|      1|     4|
| 15|2024-12-05|2024-12-05 04:19:...|      15|      1|     5|
| 16|2024-12-05|2024-12-05 04:19:...|      16|      1|     6|
| 17|2024-12-05|2024-12-05 04:19:...|      17|      1|     7|
| 18|2024-12-05|2024-12-05 04:19:...|      18|      1|     8|
| 19|2024-12-05|2024-12-05 04:19:...|      19|      1|     9|
+---+----------+--------------------+--------+-------+------+
```

```py
def rows_per_partition(sdf: "DataFrame") -> None:
    num_rows = sdf.count()
    sdf_part = sdf.withColumn("partition_id", f.spark_partition_id())
    sdf_part_count = sdf_part.groupBy("partition_id").count()
    sdf_part_count = sdf_part_count.withColumn("count_perc", 100*f.col("count")/num_rows)
    sdf_part_count.orderBy("partition_id").show()
```

```py
def rows_per_partition_col(sdf: "DataFrame", num_rows: int, col: str) -> None:
    sdf_part = sdf.withColumn("partition_id", f.spark_partition_id())
    sdf_part_count = sdf_part.groupBy("partition_id", col).count()
    sdf_part_count = sdf_part_count.withColumn("count_perc", 100*f.col("count")/num_rows)
    sdf_part_count.orderBy("partition_id", col).show()
```

ここまでは、以前と同様です。

ファイルの格納場所として、Unity Catalogのボリュームを指定します。

```py
base_dir = "/Volumes/takaakiyayoi_catalog/spark/data"
```

100万レコードのデータフレームを保存します。

```py
num_rows = 1000000
```

# 1. パーティション数に基づいて、いくつのファイル数が書き込まれるのか

- 答えは簡単です。ParquetはSparkのパーティションに基づいてデータを複数のsnappyファイルに分割します。これにより、これらのファイルを並行して書き込むことができます。
- Parquetの詳細はこちらで確認できます: https://learncsdesigns.medium.com/understanding-apache-parquet-d722645cfe74

パーティション数を1にして、Parquetとして保存します。

```py
sdf = sdf_generator(num_rows, 1)
print(sdf.rdd.getNumPartitions())
sc.setJobDescription("Write 1 file")
sdf.write.format("parquet").mode("overwrite").save(f"{base_dir}/test_1_file.parquet")
sc.setJobDescription("None")
```
```
1
```

カタログエクスプローラで保存先を確認すると、一つのsnappy.parquetファイルが保存されていることを確認できます。

![1.snappy.parquet.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/47ab473b-ea48-41b7-6d6a-8cba12689ea0.png)

4パーティション、12パーティションで保存します。

```py
sdf = sdf_generator(num_rows, 4)
print(sdf.rdd.getNumPartitions())
sc.setJobDescription("Write 4 file")
sdf.write.format("parquet").mode("overwrite").save(f"{base_dir}/test_4_file.parquet")
sc.setJobDescription("None")
```
```
4
```

![4.snappy.parquet.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/7af4e2d4-32bd-84c0-34b1-43f22b7ddb54.png)

```py
sdf = sdf_generator(num_rows, 12)
print(sdf.rdd.getNumPartitions())
sc.setJobDescription("Write 12 file")
sdf.write.format("parquet").mode("overwrite").save(f"{base_dir}/test_12_file.parquet")
sc.setJobDescription("None")
```
```
12
```
![12.snappy.parquet.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/df1d6592-67c1-c39b-26c7-3d0fc440813b.png)


# 2. Spark UIでは何が表示されますか？
- 各書き込みに対して1つのジョブと1つのステージがあります。
- 書き込みにおいても、カウントがパーティションに分割する方法を決定します。1つのパーティションは、4コアにおいて4または12のパーティションよりも少ないです。12のパーティションは8MBのデータを書き込む際にオーバーヘッドがあるようです。
- SQLタブでは、書き込まれたファイルの数とデータ量も確認できます。
- ステージでは、書き込まれた行数とサイズ、実行時間も確認できます。

パーティション数が増えると、書き込みに多いても並列化の恩恵を受けることができ、書き込みを高速化することができます。

![writing jobs.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/2626d78f-f8cf-5c2a-7f6f-db88e5adb1f6.png)

**1パーティション**

![1 partition writing.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/65b2bb60-b0f9-0012-1629-f2cdd33b0fc6.png)

**4パーティション**

![4 partition writing.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c48e18e2-a067-f977-bd35-46e944f28478.png)

**12パーティション**

![12 partition writing.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/167025bc-33b7-3015-222c-55e3c1f58d47.png)
![12 partition writing sql.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/dc552f72-8c27-fe22-7b14-3a7c5d48f2f3.png)

# 3. Coalesce と Repartition を使用してデータを保存する
- **注意:** repartition はコストのかかる操作ですが、後続のプロセスに役立つ場合があります。
- 1つのパーティションに減らす場合、`repartition(1)` は `coalesce(1)` よりも優れていることがあります。すべてのコアが処理ステップに使用されるためです。

```py
sdf = sdf_generator(num_rows, 12)
sdf = sdf.coalesce(4)
print(sdf.rdd.getNumPartitions())
sc.setJobDescription("Write with Coalesce 12 to 4 file")
sdf.write.format("parquet").mode("overwrite").save(f"{base_dir}/test_12_to_4_coalesce_file.parquet")
sc.setJobDescription("None")
```
```
4
```

![Write with Coalesce 12 to 4 file.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c181bbd9-cabd-eca0-d091-d3608f40bd0f.png)

```py
sdf = sdf_generator(num_rows, 12)
sdf = sdf.coalesce(1)
print(sdf.rdd.getNumPartitions())
sc.setJobDescription("Write with Coalesce 12 to 1 file")
sdf.write.format("parquet").mode("overwrite").save(f"{base_dir}/test_12_to_1_coalesce_file.parquet")
sc.setJobDescription("None")
```
```
1
```

![Write with Coalesce 12 to 1 file.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/f69d7419-eca1-e4e3-5c31-d773dee70602.png)

```py
sdf = sdf_generator(num_rows, 12)
sdf = sdf.repartition(4)
print(sdf.rdd.getNumPartitions())
sc.setJobDescription("Write with Repartition 12 to 4 file")
sdf.write.format("parquet").mode("overwrite").save(f"{base_dir}/test_12_to_4_repartition_file.parquet")
sc.setJobDescription("None")
```
```
4
```


![Write with Repartition 12 to 4 file.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c68db190-9bcf-b80b-b773-2501991ab422.png)

```py
sdf = sdf_generator(num_rows, 12)
sdf = sdf.repartition(1)
print(sdf.rdd.getNumPartitions())
sc.setJobDescription("Write with Repartition 12 to 1 file")
sdf.write.format("parquet").mode("overwrite").save(f"{base_dir}/test_12_to_1_repartition_file.parquet")
sc.setJobDescription("None")
```
```
1
```

![Write with Repartition 12 to 1 file.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/ef83eaef-8422-d19d-fc53-1e5575cd63f1.png)

# 4. 書き込み時の空のパーティションの問題

- Sparkは、例えば、フィルタリング後に実際のレコード/データが含まれているパーティションのみを書き込むほどに十分スマートです。

```py
sdf = sdf_generator(num_rows, 20)
sdf = sdf.filter(f.col("id") < 200)
print(sdf.rdd.getNumPartitions())
sc.setJobDescription("Empty rows")
sdf.write.format("parquet").mode("overwrite").save(f"{base_dir}/emptyRows.parquet")
sc.setJobDescription("None")
```
```
20
```
```py
rows_per_partition(sdf)
```

```
+------------+-----+----------+
|partition_id|count|count_perc|
+------------+-----+----------+
|           0|  200|     100.0|
+------------+-----+----------+
```

このデータフレームは20パーティションですが、レコードを含む1ファイルしか保存されていません。

![emptyRows.parquet.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/8b987ef2-4501-77e6-4b2d-c82ee8597af1.png)

# 5. idfirst列でのrepartition

[以前](https://qiita.com/taka_yayoi/items/86cd34b1c5617a5b8fd6#2-%E3%83%AA%E3%83%91%E3%83%BC%E3%83%86%E3%82%A3%E3%82%B7%E3%83%A7%E3%83%8B%E3%83%B3%E3%82%B0%E3%81%AE%E4%BB%95%E7%B5%84%E3%81%BF)と同様に、偏りのある`idfirst`で、かつ少ない数でのrepartitionを行うと、同じパーティションに複数の`idfirst`のデータが含まれ、空のパーティションが生成されることになります。この場合は、データを持たない空のパーティションが保存されます。

```py
sdf = sdf_generator(num_rows, 20)
sdf = sdf.repartition(10, "idfirst")
print(sdf.rdd.getNumPartitions())
sc.setJobDescription("repartition 10 idfirst")
sdf.write.format("parquet").mode("overwrite").save(f"{base_dir}/repartition_10_idfirst.parquet")
sc.setJobDescription("None")
```
```
10
```

```py
rows_per_partition(sdf)
```
```
+------------+------+----------+
|partition_id| count|count_perc|
+------------+------+----------+
|           3|111111|   11.1111|
|           4|111111|   11.1111|
|           5|111112|   11.1112|
|           6|222222|   22.2222|
|           8|111111|   11.1111|
|           9|333333|   33.3333|
+------------+------+----------+
```

上で表示されていないパーティションIDのファイルをカタログエクスプローラで特定し、以下で中身を表示してみます。

```py
spark.read.parquet(f"/Volumes/takaakiyayoi_catalog/spark/data/repartition_10_idfirst.parquet/part-00000-tid-4034718789270274402-0129652f-22d3-43e1-a61a-2642b8aaf027-341-1-c000.snappy.parquet").show()
```
```
+---+----+---------+--------+-------+------+
| id|date|timestamp|idstring|idfirst|idlast|
+---+----+---------+--------+-------+------+
+---+----+---------+--------+-------+------+
```

パーティション数を変えて挙動を確認してみましょう。

```py
sdf = sdf_generator(num_rows, 20)
sdf = sdf.repartition(8, "idfirst")
print(sdf.rdd.getNumPartitions())
sc.setJobDescription("repartition 8 idfirst")
sdf.write.format("parquet").mode("overwrite").save(f"{base_dir}/repartition_8_idfirst.parquet")
sc.setJobDescription("None")
```
```
8
```

```py
rows_per_partition(sdf)
```
```
+------------+------+----------+
|partition_id| count|count_perc|
+------------+------+----------+
|           0|111111|   11.1111|
|           1|111111|   11.1111|
|           2|222222|   22.2222|
|           3|333334|   33.3334|
|           5|111111|   11.1111|
|           6|111111|   11.1111|
+------------+------+----------+
```

こちらに続きます。

https://qiita.com/taka_yayoi/items/6199b0104f51066ba5d2

### はじめてのDatabricks

[はじめてのDatabricks](https://qiita.com/taka_yayoi/items/8dc72d083edb879a5e5d)

### Databricks無料トライアル

[Databricks無料トライアル](https://databricks.com/jp/try-databricks)