はじめに
業務でAWS Glueを触る機会があり、
『Glue Job実行でデータ重複を避ける方法』で困ったので備忘録として残しておきたいと思います。
対象ケース
- GLue Jobでデータソースから定期的に最新データを取得したい
以下具体例です。
RDSからGlue Jobで最新データを取得し、S3に保存。
保存データはQuickSight、Athena、SageMakerなどに使用する。
本題
今回の目的である
「Glue Job実行時にデータ重複を避ける」方法は実はいくつかありますが、
今回ご紹介するもの以外の方法は、
一定の制約があるものや、SQL文の追加が必要だったりして少し複雑です。
(と自分は思っちゃいました。自分の技術不足なだけかもしれませんが。。。)
原因
データ重複が起きる原因としては、前回Job実行時のデータがS3に残っているということです。
思いついた解決方法
Glue Job実行のタイミングでS3のデータ全消しして丸ごと最新データに置き換えればいいじゃん！！
図に起こすとこんな感じです。
1.まずはGlue JobがS3バケットにアクセスして過去データを消去します。
2.Glue JobがRDSにアクセスしデータを取得、
取得データを空になっているS3バケットに保存する。
これでS3内はRDSから取得したての最新データのみになり、
その後の接続先からも最新データを扱えるということになるわけです。
手順
前置きが長くなりましたが、実際の手順をご紹介します。
タイトル通りめっちゃ簡単です。
Glue Jobスクリプトにpurge_s3_pathの１行を追加するだけです！
1.Glue Jobのスクリプト編集ページを開きます。
2.スクリプト内から以下コードを見つけてください。
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
この部分に
purge_s3_path(s3_path, options={}, transformation_ctx="")
を追記してください。
作業はこれで完了です。
ちなみに
purge_s3_pathのパラメータについてはこんな感じです。(公式ドキュメントから引用)
s3_path – 削除するファイルの Amazon S3 のパス（s3://// 形式）
-
options – 削除するファイルとマニフェストファイルの生成のためのファイルをフィルタリングするオプション。
- retentionPeriod – ファイルを保持する期間を時間単位で指定します。保存期間より新しいファイルは保持されます。デフォルトでは 168 時間（7 日）に設定されています。
- partitionPredicate – この述語を満たすパーティションは削除されます。これらのパーティションの保存期間内のファイルは削除されません。デフォルトでは "" – 空に設定されます。
- excludeStorageClasses – excludeStorageClasses セット内のストレージクラスを持つファイルは削除されません。デフォルトは Set() – 空白のセットです。
- manifestFilePath – マニフェストファイルを生成するためのオプションのパス。正常にパージされたすべてのファイルが Success.csv に記録され、失敗したファイルは Failed.csv に記録されます。
transformation_ctx – 使用する変換コンテキスト (オプション)。マニフェストファイルパスで使用されます。
catalog_id – アクセス中の データカタログ のカタログ ID（データカタログ のアカウント ID）。デフォルトでは、None に設定されています。None のデフォルト値は、サービス内の呼び出し元アカウントのカタログ ID になります。
最後に
今回はGlue Job実行時のデータ重複を避ける方法についてめっちゃ簡単な方法をご紹介しました。
どなたかのお役に立てていれば幸いです。
お読みいただきありがとうございました。