Snowpipeについて調べる。自動ロードや差分ロードが良い感じに出来るのか

Snowpipeとは

https://docs.snowflake.com/ja/user-guide/data-load-snowpipe-intro

COPYコマンドとかとは異なり、ファイル単位で小さい粒度のロードが出来るということらしい。

用語の整理

• Snowpipe→機能名
• パイプ→名前付きファーストクラスSnowflakeオブジェクト
• Snowflakeテーブル→Snowflakeの世界のテーブル
• 内部ステージ→Snowflakeのテーブルとか、名前付きステージ
• 外部ステージ→S3とかのファイルの置き場所のこととか
• 半構造化データ→JSONとかAVROとか

使い方

パイプを作ることでなんやかんやするらしい

CREATE PIPE ALTER PIPE DROP PIPE DESCRIBE PIPE SHOW PIPES

一旦CREATE PIPEさえ理解すれば、使い始められそうではある。

基本、RDBでいうところのTABLEがPIPEに置き換わった構文のようだ。

CREATE PIPE

https://docs.snowflake.com/ja/sql-reference/sql/create-pipe

こう書くらしい。

create pipe mypipe_s3
  auto_ingest = true
  aws_sns_topic = 'arn:aws:sns:us-west-2:001234567890:s3_mybucket'
  as
  copy into snowpipe_db.public.mytable
  from @snowpipe_db.public.mystage
  file_format = (type = 'JSON');

このように、ステージからsnowflakeテーブルへと、データを流し込むことになる。
データがS3にputされる→SNS経由でSQSにキュー送信→Snowflake側でサブスクライブ→PIPEの定義通り、ステージからテーブルにCOPYという流れが組まれるということらしい。

ややこしいのは、Snoflake側がSQSを作成、管理しており、S3側がその送信先を知っていないといけないこと。

このあたり、日本語ドキュメントがよく分からなかったので結局英語の公式ドキュメントを読むことになった

英語版 https://docs.snowflake.com/en/user-guide/data-load-snowpipe-auto-s3

日本語版 https://docs.snowflake.com/ja/user-guide/data-load-snowpipe-auto-s3

差分ロードなどの仕組み

上記のようなイベント通知を使った自動ロードを使えば、差分を日時で取り込むワークフローなども作りやすそうである。
S3以外にもCloud StorageやAzure Blobなどメインどころのクラウドのストレージサービスに対応してる。

全量をドカンとロードしたあと、日時データをS3にアップロードし、毎日S3に1日分をputしてサブスクライブ→PIPEで差分をロードとか出来そう。

あとは、REST APIも用意されてるらしいので
それでよしなに自分でロードするためのバッチでも作るのも出来そう。

取り込む内容についてはfile formatが対応していればゴニョゴニョ出来そう。
例えば「csvファイルの特定ファイルだけSELECTしてCOPYする」など

https://docs.snowflake.com/en/user-guide/querying-stage

まとめ

• SnowpipeでS3からSnowflakeへのファイル単位のロードが出来る
• PIPEのDDLでSnowpipeを構築できる（CREATE PIPEなど）
• SQSのサブスクライブかRESTのAPIコールによってPIPEを動かすことが出来る
• ロードの単位はファイルっぽい。ファイルある程度SELECT文で条件つけて取り込むことも出来るが、あまり難しいことは出来ない（COPY構文の範囲内のことしか出来ない）