Glueで特定のパーティションのみ上書きする方法（動的パーティション上書きモード）

# やりたいこと
Glueで追加のデータを書き込む際に、以下のような書き込みがしたいことがありました。
- 追加データに含まれるパーティション内は上書き
- 含まれないパーティションはそのまま

## 具体例
以下2つのデータを考えます。
origin.csvが既存のデータで、add.csvが追加のデータです。
```origin.csv
date,item,num,price
20230301,A,3,100
20230301,B,1,90
20230301,C,2,70
20230302,D,5,140
20230302,E,7,210
20230303,F,3,500
```

```add.csv
date,item,num,price
20230302,G,5,130
20230303,H,2,85
20230304,I,3,440
```
この場合、以下のような挙動を期待しています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2949357/92431ddd-35b7-2a4e-02bf-1b6107c66d54.png)
- date=20230301は追加のデータに含まれないためそのまま残る
- 既存のデータにも存在したdate=20230302, 20230303の2つのパーティション内は上書き
- 既存のデータに存在しないdate=20230304のパーティションは追加

これは、通常の上書き（overwrite）では実現できません。
overwriteをすると、以下のような挙動になってしまいます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2949357/4dfbcdf6-6c2e-99d6-8ec7-f714fd7b46ed.png)
- 追加のデータのdate=20230302, 20230303, 20230304が上書きされる
- date=20230301は追加のデータに含まれないため削除されてしまう

# 前提条件
Glue version3.0を利用しています。
[Sparkのversion](https://docs.aws.amazon.com/ja_jp/glue/latest/dg/add-job.html#:~:text=%E3%81%A6%E3%81%8F%E3%81%A0%E3%81%95%E3%81%84%E3%80%82-,AWS%20Glue%20%E3%83%90%E3%83%BC%E3%82%B8%E3%83%A7%E3%83%B3,-AWS%20Glue%20%E3%81%AE)は3.1.1です。

# 方法
Sparkの[動的パーティション上書きモード](https://docs.aws.amazon.com/ja_jp/emr/latest/ReleaseGuide/emr-spark-committer-reqs.html#:~:text=%E4%BE%8B%20%2D-,%E5%8B%95%E7%9A%84%E3%83%91%E3%83%BC%E3%83%86%E3%82%A3%E3%82%B7%E3%83%A7%E3%83%B3%E4%B8%8A%E6%9B%B8%E3%81%8D%E3%83%A2%E3%83%BC%E3%83%89,-%E4%BB%A5%E4%B8%8B%E3%81%AE%20Scala)を使って書き込むと、実現できます。
このモードは、Spark2.4から登場したそうです。Glueジョブversion3.0はSpark3.1.1なので利用できますね。

以下、サンプルコードです。
```python
import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

### 処理ここから
input_s3_path="s3://test-bucket/input/add/"
output_s3_path="s3://test-bucket/output/"

# input
df = spark.read.csv(input_s3_path, header=True)

# output
df.write.mode("overwrite")\
  .option("partitionOverwriteMode", "dynamic")\
  .partitionBy("date")\
  .parquet(output_s3_path)
### 処理ここまで

job.commit()
```

## 実際に実行してみる
上で出した例のcsvを用意して、まずは準備します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2949357/37fc2960-0dec-0c24-c4db-04375232f492.png)
それぞれのパーティションの中をS3 Selectで確認します。

**date=20230301**
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2949357/b7c07d89-e5d5-484c-da1f-8f7ddf90042e.png)

**date=20230302**
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2949357/6f592fb8-d2a1-eb8a-1718-27095aa2546f.png)

**date=20230303**
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2949357/0a3b9e3e-d0ef-308d-8332-fc127f997cf4.png)

上記のソースでGlueジョブを実行します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2949357/8d8cdecc-9f5a-ab9f-0021-595e97cd7e6c.png)

期待通り、date=20230301は消えていません。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2949357/d45dd054-6a81-cecd-4d5e-f6024b4c289e.png)

同じく、それぞれのパーティションの中をS3 Selectで確認します。
**date=20230301**
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2949357/d30fa3bd-958e-cb80-dc37-745b5e60c54d.png)

**date=20230302**
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2949357/a81de74d-e672-990b-e650-d1be29f9e3b7.png)

**date=20230303**
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2949357/769df9f0-379f-2865-999a-48a1539ea46f.png)

**date=20230304**
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2949357/270dbe9f-acab-d725-5ee1-8427f7c125e4.png)

見事に、想定通りの出力になっていますね！

# まとめ
通常の上書きでは、追加データにないパーティションのデータが消えてしまいましたが、動的パーティション上書きモードを使えば消えずに存在するパーティションだけを上書きできました。
モードの指定方法は、writeの際に`option("partitionOverwriteMode", "dynamic")`と指定すればOKです。
これで、冪等性を担保して書き込みができますね！
ただし、同じパーティションに別のジョブからも書き込みがある場合は、動的パーティション上書きモードを利用すると他のジョブが書いたデータを上書きしてしまう可能性があるので注意してください。