【データ基盤構築/Snowflake】S3にエクスポートしたparquet形式のファイルをSnowflakeにロードする

# 今回の課題
前回の記事で、RDSスナップショットをS3にparquet形式でエクスポートしたので、
そのデータをSnowflakeにロードしたい。

↓前回の記事

https://qiita.com/Ayumu-y/items/523f700960129c57e3b3

## 実装手順
S3のparquetファイルをSnowflakeにロードした手順を記載していく。
前回の続きなので、

### 1)「Snowflakeにエクスポートするデータを格納しているS3バケット」にアクセス可能なIAMポリシーを作成する
AWSのIAMコンソールにて、下記でポリシーを作成する。

```json
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "s3:ListBucket",
                "s3:GetBucketLocation"
            ],
            "Resource": "arn:aws:s3:::*",
            "Condition": {
                "StringLike": {
                    "s3:prefix": [
                        "*"
                    ]
                }
            }
        },
        {
            "Sid": "VisualEditor1",
            "Effect": "Allow",
            "Action": [
                "s3:PutObject",
                "s3:GetObject",
                "s3:GetObjectVersion",
                "s3:DeleteObject",
                "s3:DeleteObjectVersion"
            ],
            "Resource": [
                "[アクセスしたいバケットのARN]",
                "[アクセスしたいバケットのARN]/*"
            ]
        }
    ]
}
```
### 2)IAMロールを作成して、そのロールに1)で作成したポリシーをアタッチする
下記の画像のように、
`信頼されたエンティティタイプ`を`AWSアカウント`に設定して、新しくロールを作成し、
1)で作成したポリシーをアタッチする。

※画像引用先：[Snowflakeで外部ステージを作成する](https://qiita.com/KimiyukiMuramatsu/items/56dbb35f48b69b7b36a8)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/883573/c4bd0d03-d37a-c5e3-dd5a-af1199956f4c.png)


### 3)Snowflakeにてストレージ統合を作成する
ストレージ統合とは、クラウドストレージとSnowflakeを接続するためのオブジェクト。
クラウドストレージにアクセスする際には、「ストレージ統合」を経由してアクセスする。
単一のストレージは、複数の外部ステージをサポートできる。

```sql
-- 外部ストレージ統合
CREATE or replace STORAGE INTEGRATION INT_S3_ACCESS
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = S3
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = [2で作成したロールのARN]
  STORAGE_ALLOWED_LOCATIONS = ('s3://[バケット名]')
;
```

こちらのクエリを実行することで**ストレージ統合**が作成された。
この「ストレージ統合」を経由して、外部のクラウドストレージにアクセスすることができるようになった。




### 4)Snowflakeが使用するAWSの「IAMユーザー」と「外部ID」を取得する
Snowflakeのworksheetにて、下記のクエリを実行して抽出されたデータから、
`STORAGE_AWS_IAM_USER_ARN`、`STORAGE_AWS_EXTERNAL_ID`のproperty_valueの値を覚えておく。

```sql
desc integration [ストレージ統合の名称];
```
ここで確認した`STORAGE_AWS_IAM_USER_ARN`、`STORAGE_AWS_EXTERNAL_ID`のproperty_valueの値を、AWSのIAMロール作成時に利用する。


### 5)2)で作成したロールの信頼ポリシーを編集する
そして次に、ロールの`信頼ポリシーを編集`をクリックして、
3)にて`desc integration`で取得した、`STORAGE_AWS_IAM_USER_ARN`と`STORAGE_AWS_EXTERNAL_ID`の値を下記に追記する。
```json
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "AWS": "[desc integrationで取得したSTORAGE_AWS_IAM_USER_ARNの値]"
            },
            "Action": "sts:AssumeRole",
            "Condition": {
                "StringEquals": {
                    "sts:ExternalId": "[desc integrationで取得したSTORAGE_AWS_EXTERNAL_IDの値]"
                }
            }
        }
    ]
}
```


### 6)ファイルフォーマットの作成
下記のクエリで、Snowflakeのテーブルにアクセスまたはロードするための、
ファイルの形式を事前に定義しておく。

今回はparquet形式のファイルをSowflakeにロードしたいので、
`type = parquet`と定義している。
```sql
create or replace file format parquet_format
    type = parquet
;
```

### 7)外部ステージを作成
下記のクエリで外部ステージを作成する。
これによって、Snowflakeから外部ステージのデータに対してクエリが実行できるようになった。
```sql
create or replace stage external_stage
    url = 's3://[バケット名]'
    storage_integration = INT_S3_ACCESS
    file_format = parquet_format
;
```

下記のようなクエリで、この外部ステージのファイルに対してクエリを実行することができる。
FROM句に、`@[外部ステージ名]/[S3のファイル名]/[S3のファイル名]`と記述することで、S3内のファイルにアクセスしてクエリでデータを抽出することができる。
```sql
select $1 from @EXTERNAL_TABLE/mysnapshot20230402043959/rails_test_app_test/rails_test_app_test.purchase_log/1/part-00000-cc307475-00bf-4e8f-a351-b358a1636058-c000.gz.parquet
(file_format => parquet_format)
```
そして、上記のクエリを実行することで、下記の画像のようにparquet形式のデータを抽出することができる。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/883573/98f9c92d-e2e1-9e1b-69fc-2a52a3b15b30.png)

※上記のクエリで呼び出せない場合は、rdsからS3にスナップショットをエクスポートする際にKMSで暗号化していることが原因かもしれないので、[こちら](https://qiita.com/Ayumu-y/items/7ef351524c48d853f301#3-1s3%E3%83%90%E3%82%B1%E3%83%83%E3%83%88%E3%81%AB%E5%AF%BE%E3%81%99%E3%82%8B%E6%A8%A9%E9%99%90%E3%81%AE%E3%83%9D%E3%83%AA%E3%82%B7%E3%83%BC%E3%82%92%E4%BD%9C%E6%88%90%E3%81%99%E3%82%8B)を確認してください。






### 8)テーブルを作成する
S3のparquet形式のファイル（=RDSのスナップショット）を格納するためのテーブルを作成しておく。

```sql
-- 購入履歴のようなテーブルを今回はサンプルで用意した。
create table DEMO_DB.PUBLIC.purchase_log(
    dt varchar(255)
    , order_id integer
    , user_id varchar(255)
    , purchase_amount integer
)
;
```
### 9)8)で作成したテーブルにS3のデータを格納する
下記のクエリで、parquet形式のファイルをテーブルにきれいに収まるように調整して、
8)で作成したテーブルにデータを格納する。
```sql
copy into purchase_table
from
(
SELECT 
    $1:dt::varchar
    , $1:order_id::varchar
    , $1:user_id::varchar
    , $1:purchase_amount::varchar
FROM
    @EXTERNAL_TABLE/mysnapshot20230402043959/rails_test_app_test/rails_test_app_test.purchase_log/1/part-00000-cc307475-00bf-4e8f-a351-b358a1636058-c000.gz.parquet
(file_format => parquet_format)
)
;
```
下記のクエリでを実行すると、

```sql
SELECT 
    $1:dt::varchar
    , $1:order_id::varchar
    , $1:user_id::varchar
    , $1:purchase_amount::varchar
FROM
    @EXTERNAL_TAB/mysnapshot20230402043959/rails_test_app_test/rails_test_app_test.purchase_log/1/part-00000-cc307475-00bf-4e8f-a351-b358a1636058-c000.gz.parquet
(file_format => parquet_format)
;
```
下記の画像のように、parquet形式のデータをテーブルの形に整形することができる。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/883573/caa4363b-4983-0476-3eac-90f3e3e24a1e.png)

# まとめ
以上で、S3からSnowflakeにデータをロードすることができた。


#### 参考記事

https://zenn.dev/kyami/articles/40771c49741c01

https://qiita.com/KimiyukiMuramatsu/items/56dbb35f48b69b7b36a8