公式ドキュメントを見ながらポチポチしていくとだいたいロード出来てしまうのだが、
公式ドキュメントを良く読んだことがなかったのでなぞってみる。
いくつかあるロード方法のうちストレージ統合を利用し実際にS3からロードしてみた。
Snowflakeへのデータのロード
https://docs.snowflake.com/ja/user-guide-data-load.html
【目次】
- ∨ステージ
- ∨COPY INTOによるバルクロード
- ∨Snowpipeを使用した連続ロード
- ∨外部にあるデータをロードせずにクエリ実行
- ∨ファイル形式と半構造化データ
- ∨圧縮形式
- ∨名前付きファイル
- ∨クラウドストレージの認証情報をどう持つべきか
- ∨[ストレージ統合オブジェクト版] S3からSnowflakeにデータをロードしてみる
- ∨まとめ
ステージ
gitで言うステージとほぼ同様の概念で、ローカルにあるデータをSnowflakeのテーブルに
アップロードする前に一旦Snowflakeが管理する領域にアップロードすることができる。
Snowflakeが管理する領域の外にあるか内にあるかで外部ステージ/内部ステージに分かれている。
| 名称 | 概要 |
|---|---|
| 外部ステージ | Snowflakeの外のストレージ。 例えば主要3クラウドサービスのおブロックストレージ AWS S3,Google Cloud Storage,Azure Blob Glacier Deep Archiveなど,そのままでは使えず復元操作が必要なものは対象外。 これらに名前をつけて扱えるようにする(名前付き外部ステージ)。 例えば名前付きステージとして設定したS3に対して s3 syncコマンドでローカルからコピーし,そこからSnowflakeにロードする。 |
| 内部ステージ | Snowflake側のストレージ。 ユーザーステージ ファイルを保存するために各ユーザに割り当てられているストレージ。 ユーザは内部ステージに置いたファイルを複数のテーブルにロードする。 テーブルステージ Snowflake上にあるテーブルごとに紐づいたストレージ。 複数のユーザで共有する。ここにファイルを置いてテーブルにロードする 名前付き内部ステージ テーブルとは独立して用意される複数人共有可能なストレージ。 |
各ブロックストレージが用意する操作により外部ステージにアップロードできる。
PUTコマンドによりローカルから内部ステージに直接アップロードすることもできる。
外部/内部ステージいずれのステージからもCOPY INTOコマンドでSnowflakeテーブルにコピーする。
COPY INTOによるバルクロード
例えばデータファイルが1TBとかあるとステージングも大変だしテーブルへのロードも大変。
どうにかしてステージに上げたとして、COPY INTOコマンドでテーブルにロードすることを考える。
クエリがCOPY INTOのソースとなるが、クエリをいじくることでCOPY INTOする列を変更できる。
省略したり削除したり名前を変えたりキャストしたり、クエリ操作で出来ることが出来る。
1TBのステージをCOPY INTOでロードすると、仮想ウェアハウスがむちゃくちゃ頑張る。
仮想ウェアハウスを盛れば盛るほど信じられない速度でCOPY INTOが完了して気分が良い。
Snowpipeを使用した連続ロード
少量のデータを段階的にロードする方法。断続的にパイプのこちら側から向こう側へデータを流す。
COPY INTOとは異なりSnowflakeが用意するクラウドリソースの課金となる。(不明..)
COPY INTOと同様のデータ変換処理をかけられる。(不明..)
外部にあるデータをロードせずにクエリ実行
Snowflakeのテーブルにロードせず、外部のクラウドストレージに置いたままクエリを実行できる!
ただ外部のクラウドストレージにあるだけではダメで「外部テーブル」を設定する。
外部テーブルにより外部のクラウドストレージにメタ情報が設定される仕組み。
外部テーブル上のデータにはDMLを実行できない。
とはいえ、外部テーブルにあるままクエリをかけると遅い。
これを解消するのがMaterialized Viewのシナリオで、外部テーブルをオリジンとして
Materialized Viewを作っておけば、外部テーブルの変更に対して透過的にアクセスできるし、
クエリ結果キャッシュほどでは無いにせよ、パフォーマンス上の利点がある。
もはやデータのロードではないので、新たに記事をおこして書く。
ファイル形式と半構造化データ
外部ステージに指定されたファイルタイプのファイルを置くことでSnowflakeのテーブルにロードできる。
通常、CSVやTSVなんかを置くイメージ。
特筆すべきはJSONのようにスキーマ定義がコンテナと同一化しているフォーマットを「半構造化データ」
として読み込むことができるということ。JSON以外にHive由来のArvo,ORC,Parquetに対応している。
(半構造化データについては奥が深そうなので別記事対応。)
| フォーマット名称 | |
|---|---|
| Arvo | 要はシリアル化されたデータとデータのスキーマを同時に格納するフォーマット。 Big Queryも対応しているしこの界隈では有名なんだろうか。 Apache Arvo |
| ORC | Apache Hive用の列指向フォーマット。 The Optimized Row Columnar (ORC) file format provides a highly efficient way to store Hive data. It was designed to overcome limitations of the other Hive file formats. Using ORC files improves performance when Hive is reading, writing, and processing data. the smallest, fastest columnar storage for Hadoop workloads. Apache ORC |
| Pqrquet | databricksからApache Pqrquet。列志向フォーマット。 Apache Parquet は、効率的なデータの保存と検索のために設計された、オープンソースの列指向データファイル形式です。複雑なデータを一括処理するための効率的なデータ圧縮と符号化方式を提供し、パフォーマンスを向上させます。 |
ファイルは基本的にはUTF8。ただASCIIで済むのにUTF8はちょっと贅沢だなと思ったりもする。
圧縮形式
ローカルで.gzに圧縮した後ステージに載せていたが圧縮していないファイルをステージに載せようとすると
自動的にgzip圧縮がかかる。特に圧縮形式にこだわりがないなら自動gzip化で良さそう。
公式には圧縮アルゴリズムが書いてあるが,要はzip,gzip,bz2。
ちなみに、ファイルのロードを指示する際、拡張子を省略した書き方ができるが、
同じファイル名で圧縮形式が違うファイルを同一階層に置いておくと、片方しか読まれない。
ファイルを更新したのにロード操作をしても全然反映されない、とハマる。
名前付きファイル
現物のファイルを上位概念化したオブジェクト。ファイルのメタデータをまとめるために存在する。
例えば現物のファイルをアップロードしようとすると、ローカルのパスとかアクセスのための権限など
様々な情報が必要となる。もしそのファイルを何度も使うとすると毎回同じ内容を書かないといけない。
必要なメタデータをまとめておいてそれを使うことで、手間を省略できる。
クラウドストレージの認証情報をどう持つべきか
例えばS3を外部ステージとして設定することができるが、ではS3の認証情報をどう持つべきか。
以下に3つの選択肢が書かれている。
Amazon S3へのセキュアアクセスの構成
https://docs.snowflake.com/ja/user-guide/data-load-s3-config.html
- [推奨]ストレージ統合オブジェクト
- [廃止]AWS IAM ロール
- [1回限り]AWS STSによる一時認証
ストレージ統合オブジェクト
通常、運用者がアクセスキーとシークレットを触れない場合は多いだろうから、
管理者にストレージ統合オブジェクトの作成を依頼する、という仕組みが良さそう。
AWS IAMロール
IAMロールを払い出して利用するパターンは[廃止]となっていていずれ消える様子。
expireしない認証情報はよくない。
AWS STSによる一時認証
AWS STSによりIAMロールを有効期限付きで移譲し一時的にIAMロールアクセスできる。
IAMユーザのアクセスキー、シークレットと有効期限付きのトークンから構成される。
Snowflakeが外部ステージにアクセスする際に都度有効なトークンが必要だが、
1度限りのロードを行う用途なのであればこれも良さそう。
AWS STSについては以下。
https://docs.aws.amazon.com/ja_jp/IAM/latest/UserGuide/id_credentials_temp.html
[ストレージ統合オブジェクト版] S3からSnowflakeにデータをロードしてみる
CREATE INTEGRATIONロールが必要なため不可能なケースがあるが推奨されているのでやってみる。
gz圧縮済みのCSVを外部ステージ設定したS3にアップロード後,Snowflakeテーブルにロードする。
工程はざっくり以下の通り。
- S3にバケットを作成
- SnowflakeからS3にアクセス可能なカスタマ管理ポリシーを作成
- カスタマ管理ポリシーをアタッチしたIAMロールを作成しIAMユーザと関係させる
- ファイルフォーマットを作成する
- ストレージ統合オブジェクトを作成する
- S3に外部ステージを作成する
- IAMロールにSnowflakeとの信頼関係を追加する
- 疎通確認
- 検証用のダミーテーブル・ダミーデータを作る
- 外部ステージからSnowflakeテーブルにロードする
S3にバケットを作成
まず ikuty-s3test というバケット名のS3を用意してみた。
プライベートアクセスのみ可能で、ACLではなくIAMポリシーでアクセス制御をする設定。
SnowflakeからS3にアクセス可能なカスタマ管理ポリシーを作成
指定したカスタマ管理ポリシーでのみアクセス制御できるよう構成する。
ikuty-s3testにアクセスするカスタマ管理ポリシーを以下の通り作成する。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 |
{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "s3:PutObject", "s3:GetObject", "s3:DeleteObjectVersion", "s3:DeleteObject", "s3:GetObjectVersion" ], "Resource": "arn:aws:s3:::ikuty-s3test/*" }, { "Effect": "Allow", "Action": "s3:ListBucket", "Resource": "arn:aws:s3:::ikuty-s3test" }, { "Effect": "Allow", "Action": "s3:ListAllMyBuckets", "Resource": "*" } ] } |
カスタマ管理ポリシーをアタッチしたIAMロールを作成しIAMユーザと関係させる
IAMロールを作成し、上のカスタマ管理ポリシーをアタッチする。
IAMユーザを作成し、IAMロールと関係させる。
該当ユーザのアクセスキー、シークレットを ~/.aws/credentialsに設定しておく。
s3:ListAllMyBucketsを許可しているため、ポリシーが作用していればaws s3 lsで一覧を取得可。
(aws configureは構成済みとし省略)
|
1 2 3 4 |
$ aws s3 ls --profile s3test 2022-05-19 02:15:00 ikuty-s3test .. .. |
ファイルフォーマットオブジェクトを作成する
gz圧縮済みのCSVをファイルフォーマットオブジェクトでラップする。
デリミタはパイプ(|)、1行目はヘッダ、空フィールドはNULL扱い、圧縮はgzip。
|
1 2 3 4 5 6 7 8 9 |
--- CSVファイルフォーマットオブジェクトを作成 CREATE OR REPLACE FILE FORMAT my_csv_format TYPE = CSV FIELD_DELIMITER = '|' SKIP_HEADER = 1 NULL_IF = ('NULL', 'null') EMPTY_FIELD_AS_NULL = true COMPRESSION = gzip ; |
ストレージ統合オブジェクトを作成する
S3ストレージ統合オブジェクトを作成する。CREATE INTEGRATIONロールが必要。
|
1 2 3 4 5 6 7 |
CREATE STORAGE INTEGRATION my_s3_integration TYPE = EXTERNAL_STAGE STORAGE_PROVIDER = S3 ENABLED = TRUE STORAGE_AWS_ROLE_ARN = '<<S3へのアクセスが認可されたroleのARN>>' STORAGE_ALLOWED_LOCATIONS = ('s3://ikuty-s3test') ; |
S3に外部ステージを作成する
次にikuty-s3testバケットを外部ステージ化する。
ファイルフォーマットとストレージ統合オブジェクトがあれば指定するだけで良い。
|
1 2 3 4 5 |
create stage my_s3_stage storage_integration = my_s3_integration url = 's3://ikuty-s3test' file_format = my_csv_format ; |
IAMロールにSnowflakeとの信頼関係を追加する
IAMロールにアタッチしたカスタマ管理ポリシーだけでは不足で、IAMロールにSnowflakeとの信頼関係を設定する必要がある。
外部ステージを作った時点で外部ステージにAWS_ROLEとAWS_EXTERNAL_IDが設定される。
この2つをIAMロールの信頼ポリシーに設定する。
Snowflake側でステージを確認する。show stageでステージ一覧を得られる。
desc stageで指定したステージの詳細情報を得られる。
その際、AWS_ROLEとAWS_EXTERNAL_IDの2つを記録しておく。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
$ show stage created_on name database_name schema_name url has_credentials has_encryption_key owner comment region type cloud notification_channel storage_integration 2022-05-19 02:03:30.712 -0700 MY_S3_STAGE ikuty ikuty s3://ikuty-s3test N N ACCOUNTADMIN ap-northeast-1 EXTERNAL AWS MY_S3_INTEGRATION $ desc stage MY_S3_INTEGRATION parent_property property property_type property_value property_default STAGE_FILE_FORMAT FORMAT_NAME String "my_csv_format" STAGE_COPY_OPTIONS ON_ERROR String "ABORT_STATEMENT" "ABORT_STATEMENT" STAGE_COPY_OPTIONS SIZE_LIMIT Long STAGE_COPY_OPTIONS PURGE Boolean false false STAGE_COPY_OPTIONS RETURN_FAILED_ONLY Boolean false false STAGE_COPY_OPTIONS ENFORCE_LENGTH Boolean true true STAGE_COPY_OPTIONS TRUNCATECOLUMNS Boolean false false STAGE_COPY_OPTIONS FORCE Boolean false false STAGE_LOCATION URL String ["s3://ikuty-s3test"] STAGE_INTEGRATION STORAGE_INTEGRATION String MY_S3_INTEGRATION STAGE_CREDENTIALS AWS_ROLE String <<記録しておく>> STAGE_CREDENTIALS AWS_EXTERNAL_ID String <<記録しておく>> STAGE_CREDENTIALS SNOWFLAKE_IAM_USER String *** DIRECTORY ENABLE Boolean false false DIRECTORY AUTO_REFRESH Boolean false false |
IAMロールに設定する信頼ポリシーは以下。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
{ "Version": "2012-10-17", "Statement": [ { "Sid": "", "Effect": "Allow", "Principal": { "AWS": "<<記録しておいたAWS_ROLE>>" }, "Action": "sts:AssumeRole", "Condition": { "StringEquals": { "sts:ExternalId": "<<記録しておいたAWS_EXTERNAL_ID>>" } } } ] } |
疎通確認
基本的には「S3にアクセスするためのカスタマ管理ポリシー」と「Snowflakeとの信頼ポリシー」の2つが正しいか。
ステージ内のファイルの一覧を取得するとこれらの疎通を確認できる。
外部名前付きステージの名称の前に「@」を付けることでSnowSQLからアクセスできる。
今回作成した my_s3_stage であれば @my_s3_stage という書き方となる。
|
1 2 3 |
list @my_s3_stage ; ... |
検証用のダミーテーブル・ダミーデータを作る
ロードする先のSnowflakeテーブルを作っていなかったので作る。
CREATE TABLEでもファイルフォーマットを指定できるので良い。
|
1 2 3 4 5 6 7 |
create or replace table my_test_table ( id integer, name string, age integer ) STAGE_FILE_FORMAT = 'my_csv_format' ; |
さて、ローカルでスキーマに合うダミーデータを作って圧縮してS3に送る。
これだけ短いとgzip圧縮かけたら余計サイズが大きくなるw
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
# dummy.csvというファイルをローカルに作成 $ cd ~/ && touch dummy.csv $ echo -e "id|name|age\n1|hogehoge|10\n2|fugafuga|20\n3|foofoo|30" > dummy.csv $ cat dummy.csv id|name|age 1|hogehoge|10 2|fugafuga|20 3|foofoo|30 # gzip圧縮する $ gzip dummy.csv $ ls -la | grep dummy -rw-r--r-- 1 ikuty ikuty 75 5 02 18:14 dummy.csv.gz # Content-Typeをgzipに指定してS3にアップロード aws s3 cp dummy.csv.gz s3://ikuty-s3test/ --profile=s3test --content-encoding "gzip" --content-type "application/x-gzip" upload: ./dummy.csv.gz to s3://ikuty-s3test/dummy.csv.gz # 確認 $ aws s3 ls s3://ikuty-s3test/ --profile=s3test 2022-05-19 02:29:02 75 dummy.csv.gz |
my_s3_stage内をリストすると今上げたファイルがあることが確認できた。
|
1 2 3 4 |
list @my_s3_stage ; name size md5 last_modified s3://ikuty-s3test/dummy.csv.gz 75 *** Thu, 19 May 2022 09:29:02 GMT |
外部ステージからSnowflakeテーブルにロードする
これまでの設定が全て上手くいくとCOPY INTOで @my_s3_stage から my_test_tableへのロードが完了する。
|
1 2 3 4 5 6 |
COPY INTO my_test_table from @my_s3_stage file_format = my_csv_format ; file status rows_parsed rows_loaded error_limit errors_seen first_error first_error_line first_error_character first_error_column_name s3://ikuty-s3test/dummy.csv.gz LOADED 3 3 1 0 |
できた。
|
1 2 3 4 5 6 |
select * from my_test_table ; ID NAME AGE 1 hogehoge 10 2 fugafuga 20 3 foofoo 30 |
まとめ
Snowflakeのデータロードについて公式ドキュメントに書いてあることをなぞってみた。
外部ステージ/内部ステージを介してテーブルにデータをロードできる。
各種クラウドストレージやフォーマットなど、いくつかのステップが抽象化されていて、
複数の組み合わせについて同じ方法で対応できる様子。
そのうちストレージ統合を使い外部ステージに設定したS3からデータをロードしてみた。