おはようございます、祥です。
以前Snowpipeに関する記事を書いたのですが、TaskからSnowpipeへの移行方法がS3とGCSで異なるため、今回はそちらをまとめてみたいと思います。
💡急ぎの方向けの要約はこちら
- メッセージの蓄積開始タイミングを考慮して切り替えを実施しましょう
- S3はPIPE作成と連動していますが、GCSはPub/Sub作成と連動している関係上、別途Pub/Sub側でキューの削除対応が必要となるケースがあります
- TaskからSnowpipeへの移行にはREFRESHオプションの利用が便利です
- データロードだけで完璧な重複排除を目指すのではなく、後続の処理で重複排除できる仕組みも併せて検討しましょう
前提
TASKでのデータ取り込みからSnowpipeへの変更を想定しているため、以下の前提で以降の内容を記載します。
- データソースから毎時でデータが配置
- 日次TASKでデータ取り込みを実施
- 切り替えのタイミングでデータソース側のデータ出力の停止/再開は実施しない
- AWS、Google Cloudの設定は他社で管理されているため、PIPE作成とあわせての設定は不可（PIPE作成の前日までに設定が完了している状態）
- PIPEでの取り込み処理開始後にTASKを停止する
- 重複なくデータを取り込みたい
S3における移行のポイント
S3はPIPE作成後にメッセージが蓄積される（Snowflake管理のSQSが作成される）ため、タスク最終起動（①）からPIPE作成（②）前までに出力された未取り込みデータ（図のd,e,f）を手動取り込みする必要があります。ただ、未取り込みデータは、PIPEのREFRESHオプションを指定して起動（③）することでキューに追加され、取り込みが可能となります。
※図の参照方法
- 横軸＝時間
- 「データ出力」＝ファイル名
- 「タスク」「PIPE」「REFRESH」＝実行・作成タイミング
- 「SQS」「取り込み／未取り込みデータ」＝イベントと対象ファイル
GCSにおける移行のポイント
GCSはPub/Sub作成のタイミングでメッセージが蓄積（①）されますが、タスクによる取り込み（②）では蓄積されたメッセージに対するアクションは発生しないため、S3と同様の流れでPIPE作成を実施する（③）とデータの重複取り込みが発生してしまいます。
よって以下の流れのようにPIPE作成の前に、Pub/Subのキュー削除（③）のオペレーションを追加することで、PIPE作成時に発生していた重複取り込みが回避可能となります。
所感
- メッセージの蓄積開始タイミングを考慮することで移行に必要な対応がわかったと思います。
- データロード時の重複排除の考慮も大切ですが、後続処理での重複排除も併せて検討しましょう。
- 以下の記事が非常に参考になりましたので併せてご覧ください。
参考URL
この記事を書いた人
