こんにちは、中村です。
今回はELTツールを使わずにSnowflakeだけで処理をできるように、まずはステージに入ったファイルをTASKで変形してマート層まで作成しようと思います。
目次
概要
dbt Projects on SnowflakeはSnowflake上でdbtを開発して実行するところまで行える機能です。
詳しくはこちらの記事をご覧ください！
今回は架空のECサイトを想定して、次のような構成で作成します。
- 日ごとの購買データがS3に蓄積されていてそれをraw_salesテーブルにコピーするTASKが定時に起動する
- raw_salesテーブルと2つのマスターをJOINしてINTERMIDIATE層、MARTS層への変換をおこなうdbtのTASKが1のタスク実施をトリガーに実行される
ELTのTASK作成
前段階：外部統合の作成
このセクションではこちらの公式ドキュメントを参考に作成します。
まずはAWSでS3でバケットを作成します。
今回作成するバケットはsnowflake-staging-salesでその配下のsales配下に日次の売上データを配置する想定です。
続いてIAMポリシーを作成します。
ポリシー作成画面でJSON形式を選択し、以下の内容を入力して作成してください。
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"s3:PutObject",
"s3:GetObject",
"s3:GetObjectVersion",
"s3:DeleteObject",
"s3:DeleteObjectVersion"
],
"Resource": "arn:aws:s3:::snowflake-staging-sales/sales/*"
},
{
"Effect": "Allow",
"Action": [
"s3:ListBucket",
"s3:GetBucketLocation"
],
"Resource": "arn:aws:s3:::snowflake-staging-sales",
"Condition": {
"StringLike": {
"s3:prefix": [
"sales/*"
]
}
}
}
]
}続いて先ほどのポリシーを付与するロールを作成します。
信頼されたエンティティタイプは「AWSアカウント」を選択。
AWSアカウントは別のAWSアカウントを選択して仮に使用しているアカウントの12桁のIDを入力してください。
オプションは外部IDを要求するにチェックを入れて外部IDを適当に入力してください。
この2つは後程書き換えます。
次の画面で先ほど作成したポリシーをアタッチします。
作成したらロール画面からARNをコピーします。
ここまで出来たらSnowflakeに移動して、SQLワークシートで以下を実行してください。
--外部統合の作成
CREATE STORAGE INTEGRATION nakamura_s3_integration
TYPE = EXTERNAL_STAGE
STORAGE_PROVIDER = S3
ENABLED = TRUE
STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::********:role/use-snowflake-staging-sales' --先ほどコピーしたARN
STORAGE_ALLOWED_LOCATIONS = ('s3://snowflake-staging-sales/sales/');
--外部ステージの作成
CREATE STAGE nakamura_s3_stage
URL = 's3://snowflake-staging-sales/sales/'
STORAGE_INTEGRATION = nakamura_s3_integration;
--STORAGE_AWS_IAM_USER_ARN, STORAGE_AWS_EXTERNAL_ID確認用
DESC INTEGRATION nakamura_s3_integration;最後のDESC INTEGRATION nakamura_s3_integration;を実行したら、STORAGE_AWS_IAM_USER_ARN, STORAGE_AWS_EXTERNAL_IDのValueをコピーし、再度AWSに戻り、ロールの信頼関係を編集します。
先ほど作成したロールの信頼関係から信頼ポリシーを編集より以下を入力してください。
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "",
"Effect": "Allow",
"Principal": {
"AWS": "<snowflake_user_arn>"
},
"Action": "sts:AssumeRole",
"Condition": {
"StringEquals": {
"sts:ExternalId": "<snowflake_external_id>"
}
}
}
]
}これで外部ステージの作成は完了です。
デモデータとテーブル、dbtモデルの作成
まず今回使うデータはこちらです。
デモデータのCSV
コピーしてご利用ください。#mst_product.csv P001,商品1,1000 P002,商品2,2000 P003,商品3,3000 P004,商品4,4000 P005,商品5,5000 P006,商品6,6000 P007,商品7,7000 P008,商品8,8000 P009,商品9,9000 P010,商品10,10000#mst_customer.csv id,name,status C001,加藤美咲,ブロンズ C002,渡辺花子,ゴールド C003,渡辺直樹,ブロンズ C004,山本由美,シルバー C005,渡辺真由美,シルバー C006,山本由美,ブロンズ C007,佐藤彩,ブロンズ C008,加藤翔太,シルバー C009,小林一郎,ゴールド C010,中村翔太,シルバー C011,小林翔太,ブロンズ C012,高橋一郎,ゴールド C013,高橋花子,ブロンズ C014,佐藤花子,ゴールド C015,加藤健一,ブロンズ C016,山本健一,ブロンズ C017,渡辺美咲,ブロンズ C018,鈴木真由美,ブロンズ C019,田中一郎,ブロンズ C020,鈴木由美,ブロンズ C021,小林美咲,シルバー C022,高橋美咲,ゴールド C023,伊藤直樹,ブロンズ C024,加藤由美,シルバー C025,小林一郎,シルバー C026,佐藤一郎,ゴールド C027,田中一郎,ブロンズ C028,伊藤彩,ブロンズ C029,田中花子,ゴールド C030,田中翔太,シルバー C031,小林翔太,シルバー C032,高橋由美,ブロンズ C033,田中真由美,ブロンズ C034,中村彩,ブロンズ C035,中村太郎,ゴールド C036,山本健一,ブロンズ C037,伊藤直樹,ブロンズ C038,渡辺一郎,ゴールド C039,伊藤一郎,シルバー C040,渡辺由美,ゴールド C041,渡辺翔太,ブロンズ C042,鈴木直樹,ゴールド C043,鈴木直樹,ブロンズ C044,田中一郎,ゴールド C045,田中美咲,シルバー C046,鈴木花子,ブロンズ C047,田中由美,ゴールド C048,加藤真由美,シルバー C049,山本彩,ゴールド C050,渡辺由美,シルバー C051,山本健一,ブロンズ C052,高橋太郎,ゴールド C053,佐藤直樹,ブロンズ C054,田中由美,ブロンズ C055,小林由美,ゴールド C056,山本一郎,シルバー C057,渡辺真由美,ブロンズ C058,鈴木直樹,ブロンズ C059,加藤直樹,ブロンズ C060,山本真由美,シルバー C061,渡辺真由美,ゴールド C062,渡辺翔太,シルバー C063,中村花子,ブロンズ C064,伊藤太郎,ゴールド C065,佐藤健一,シルバー C066,加藤花子,ゴールド C067,田中健一,シルバー C068,田中真由美,ブロンズ C069,田中翔太,シルバー C070,伊藤翔太,ブロンズ C071,伊藤美咲,ブロンズ C072,田中直樹,ブロンズ C073,鈴木美咲,シルバー C074,加藤真由美,ブロンズ C075,渡辺翔太,ブロンズ C076,佐藤美咲,ゴールド C077,渡辺花子,シルバー C078,加藤由美,ブロンズ C079,田中彩,シルバー C080,加藤彩,シルバー C081,山本太郎,ゴールド C082,小林直樹,ブロンズ C083,高橋太郎,シルバー C084,小林由美,シルバー C085,加藤由美,ブロンズ C086,渡辺翔太,ブロンズ C087,伊藤花子,シルバー C088,小林花子,シルバー C089,高橋花子,シルバー C090,渡辺翔太,シルバー C091,加藤健一,ブロンズ C092,渡辺翔太,シルバー C093,渡辺彩,シルバー C094,渡辺真由美,シルバー C095,小林健一,シルバー C096,小林花子,シルバー C097,伊藤真由美,ブロンズ C098,高橋一郎,ブロンズ C099,鈴木由美,ブロンズ C100,鈴木彩,ブロンズ C101,小林花子,ブロンズ C102,高橋太郎,ゴールド C103,田中直樹,ブロンズ C104,小林美咲,ゴールド C105,小林直樹,シルバー C106,高橋健一,シルバー C107,加藤彩,ブロンズ C108,小林由美,ゴールド C109,山本翔太,シルバー C110,佐藤由美,シルバー C111,加藤彩,ブロンズ C112,鈴木真由美,シルバー C113,加藤由美,シルバー C114,小林彩,シルバー C115,佐藤彩,ゴールド C116,中村由美,ブロンズ C117,高橋彩,ブロンズ C118,田中美咲,ゴールド C119,山本健一,シルバー C120,山本彩,ゴールド C121,佐藤直樹,ブロンズ C122,佐藤健一,ブロンズ C123,田中真由美,ゴールド C124,加藤直樹,シルバー C125,加藤彩,ゴールド C126,中村一郎,シルバー C127,伊藤太郎,シルバー C128,加藤翔太,ブロンズ C129,山本彩,シルバー C130,佐藤彩,ブロンズ C131,渡辺花子,ブロンズ C132,田中一郎,シルバー C133,渡辺直樹,ブロンズ C134,高橋太郎,ブロンズ C135,中村翔太,ブロンズ C136,高橋彩,シルバー C137,渡辺直樹,ブロンズ C138,鈴木健一,ブロンズ C139,鈴木花子,ブロンズ C140,佐藤美咲,ゴールド C141,伊藤太郎,シルバー C142,加藤直樹,シルバー C143,田中美咲,ゴールド C144,佐藤由美,ブロンズ C145,中村翔太,シルバー C146,鈴木彩,シルバー C147,小林健一,ブロンズ C148,高橋真由美,ブロンズ C149,小林彩,シルバー C150,渡辺翔太,ブロンズ C151,中村美咲,ブロンズ C152,田中花子,ゴールド C153,山本美咲,ブロンズ C154,鈴木太郎,ブロンズ C155,渡辺直樹,ブロンズ C156,渡辺健一,ブロンズ C157,鈴木直樹,ブロンズ C158,中村彩,ゴールド C159,鈴木由美,シルバー C160,小林美咲,ブロンズ C161,加藤美咲,ブロンズ C162,山本太郎,ゴールド C163,伊藤健一,ブロンズ C164,鈴木一郎,ブロンズ C165,伊藤由美,ブロンズ C166,伊藤太郎,ブロンズ C167,田中翔太,ブロンズ C168,渡辺花子,ブロンズ C169,田中翔太,ブロンズ C170,高橋翔太,ブロンズ C171,加藤太郎,シルバー C172,小林一郎,ブロンズ C173,高橋翔太,ブロンズ C174,高橋真由美,シルバー C175,山本由美,シルバー C176,中村直樹,シルバー C177,渡辺真由美,ゴールド C178,田中健一,シルバー C179,伊藤直樹,ゴールド C180,渡辺花子,シルバー C181,鈴木花子,ブロンズ C182,加藤彩,ゴールド C183,伊藤翔太,ブロンズ C184,伊藤太郎,ブロンズ C185,加藤美咲,シルバー C186,鈴木翔太,ブロンズ C187,小林太郎,ブロンズ C188,山本美咲,ブロンズ C189,小林由美,ゴールド C190,高橋一郎,ブロンズ C191,鈴木翔太,ブロンズ C192,加藤直樹,ブロンズ C193,佐藤一郎,シルバー C194,中村一郎,シルバー C195,伊藤翔太,ブロンズ C196,山本真由美,シルバー C197,高橋一郎,ブロンズ C198,伊藤太郎,ブロンズ C199,中村真由美,シルバー C200,佐藤翔太,シルバー#sales_20250721.csv id,datetime,product_id,number,customer_id 1,2025-07-21 09:53:00,P005,5,C197 2,2025-07-21 02:41:00,P004,1,C101 3,2025-07-21 22:12:00,P005,5,C057 4,2025-07-21 20:01:00,P005,1,C133 5,2025-07-21 06:59:00,P008,2,C100 6,2025-07-21 23:20:00,P002,2,C167 7,2025-07-21 22:34:00,P007,2,C109 8,2025-07-21 18:42:00,P002,3,C197 9,2025-07-21 19:32:00,P003,5,C119 10,2025-07-21 15:31:00,P008,2,C067 11,2025-07-21 07:38:00,P008,2,C116 12,2025-07-21 22:35:00,P005,2,C154 13,2025-07-21 15:41:00,P002,1,C142 14,2025-07-21 15:55:00,P008,3,C196 15,2025-07-21 15:46:00,P001,2,C145 16,2025-07-21 16:12:00,P008,3,C054 17,2025-07-21 19:35:00,P004,4,C057 18,2025-07-21 19:38:00,P006,5,C162 19,2025-07-21 14:59:00,P010,3,C095 20,2025-07-21 09:27:00,P008,2,C090 21,2025-07-21 18:41:00,P005,1,C152 22,2025-07-21 03:11:00,P004,3,C147 23,2025-07-21 14:23:00,P006,5,C046 24,2025-07-21 08:01:00,P009,1,C179 25,2025-07-21 07:59:00,P008,4,C061 26,2025-07-21 03:38:00,P002,4,C029 27,2025-07-21 12:36:00,P010,5,C048 28,2025-07-21 04:31:00,P006,5,C134 29,2025-07-21 12:49:00,P007,4,C022 30,2025-07-21 01:37:00,P010,5,C171 31,2025-07-21 20:34:00,P001,4,C102 32,2025-07-21 01:44:00,P004,4,C110 33,2025-07-21 04:55:00,P010,2,C152 34,2025-07-21 06:01:00,P006,2,C155 35,2025-07-21 20:23:00,P002,4,C177 36,2025-07-21 02:23:00,P001,5,C195 37,2025-07-21 14:19:00,P006,2,C035 38,2025-07-21 21:58:00,P004,3,C129 39,2025-07-21 10:58:00,P008,2,C079 40,2025-07-21 10:27:00,P009,2,C083 41,2025-07-21 14:53:00,P005,2,C079 42,2025-07-21 19:41:00,P003,5,C138 43,2025-07-21 13:00:00,P005,4,C179 44,2025-07-21 12:18:00,P004,1,C068 45,2025-07-21 04:09:00,P007,1,C145 46,2025-07-21 10:26:00,P006,5,C126 47,2025-07-21 05:29:00,P006,3,C093 48,2025-07-21 08:59:00,P009,4,C012 49,2025-07-21 03:34:00,P004,3,C061 50,2025-07-21 21:30:00,P007,5,C090 51,2025-07-21 11:11:00,P009,3,C015 52,2025-07-21 02:53:00,P006,3,C119 53,2025-07-21 09:35:00,P008,4,C120 54,2025-07-21 05:07:00,P002,4,C043 55,2025-07-21 04:07:00,P004,1,C188 56,2025-07-21 07:11:00,P006,1,C183 57,2025-07-21 04:50:00,P010,3,C003 58,2025-07-21 13:37:00,P004,3,C070 59,2025-07-21 02:21:00,P010,5,C029 60,2025-07-21 22:07:00,P007,3,C139 61,2025-07-21 19:25:00,P009,2,C079 62,2025-07-21 11:31:00,P007,1,C081 63,2025-07-21 05:59:00,P003,4,C049 64,2025-07-21 18:58:00,P002,4,C054 65,2025-07-21 08:02:00,P006,2,C065 66,2025-07-21 13:31:00,P006,2,C149 67,2025-07-21 18:07:00,P001,2,C187 68,2025-07-21 18:50:00,P002,2,C036 69,2025-07-21 02:13:00,P001,2,C199 70,2025-07-21 14:40:00,P001,2,C073 71,2025-07-21 12:49:00,P007,3,C106 72,2025-07-21 04:23:00,P010,3,C039 73,2025-07-21 20:53:00,P006,5,C081 74,2025-07-21 20:26:00,P002,1,C083 75,2025-07-21 01:00:00,P006,4,C021 76,2025-07-21 08:25:00,P008,1,C128 77,2025-07-21 22:51:00,P010,3,C015 78,2025-07-21 17:52:00,P001,4,C052 79,2025-07-21 03:59:00,P005,1,C148 80,2025-07-21 14:39:00,P001,5,C045 81,2025-07-21 19:32:00,P009,5,C103 82,2025-07-21 04:22:00,P009,3,C131 83,2025-07-21 20:13:00,P002,3,C042 84,2025-07-21 18:43:00,P007,3,C063 85,2025-07-21 21:22:00,P001,4,C036 86,2025-07-21 20:05:00,P005,4,C090 87,2025-07-21 09:25:00,P006,4,C076 88,2025-07-21 06:27:00,P001,5,C038 89,2025-07-21 12:30:00,P006,4,C068 90,2025-07-21 21:13:00,P006,2,C067 91,2025-07-21 05:23:00,P004,2,C017 92,2025-07-21 03:47:00,P001,1,C170 93,2025-07-21 13:51:00,P007,4,C075 94,2025-07-21 14:38:00,P008,4,C041 95,2025-07-21 02:38:00,P007,3,C025 96,2025-07-21 04:48:00,P010,5,C167 97,2025-07-21 05:39:00,P003,2,C079 98,2025-07-21 15:57:00,P002,2,C103 99,2025-07-21 08:05:00,P009,4,C006 100,2025-07-21 19:25:00,P008,4,C058
これをもとにSnowflakeでテーブルを作成していきます。
--rawテーブル作成
CREATE OR REPLACE TABLE raw_sales (
id STRING,
datetime DATETIME,
product_id STRING,
number INTEGER,
customer_id STRING
);残りのテーブルはdbt runで作成できるので、続いてdbtのモデルを作成していきましょう。
ワークスペースで+ Add newからdbt Projectを選択
dbtのプロジェクト名とロール、ウェアハウス、データベース、スキーマを入力しましょう
ここまで出来たら、dbtモデルを作成していきましょう。
models配下に以下の3つのファイルを作成してください。
#source.yaml
sources:
- name: sales
schema: sales
tables:
- name: mst_product
- name: mst_customer
- name: raw_sales#int_sales.sql
{{ config(materialized='table') }}
SELECT
s.datetime,
m.product_name,
s.price,
s.number,
c.customer_name,
s.status
FROM {{source("sales","raw_sales")}} as r
LEFT JOIN {{source("sales","mst_product")}} m ON r.product_id = m.id
LEFT JOIN {{source("sales","mst_customer")}} c ON r.customer_id = c.id;#marts_sales.sql
{{ config(materialized='table') }}
select
date(datetime) as date,
status,
sum(price*number) as sales
from {{ref("int_sales")}}
group by date(datetime),status作成後、dbt run を実行し、エラーなく完了するかを確認しましょう。
実行の確認ができたら、dbt Projectをデプロイしましょう。
デプロイするデータベースとスキーマを選択し、先ほど作ったdbプロジェクトの名前を付けます。
これでテーブル作成の準備ができました。つぎはTASKを作成します。
TASKの作成
続いてSQLワークシートに戻って以下を実行してください。
#外部ステージのファイルをraw_tableにコピーするTASK
CREATE OR REPLACE TASK daily_copy_task
WAREHOUSE = nakamura_wh
SCHEDULE = 'USING CRON 0 3 * * * UTC' -- 毎日12時に実行
AS
COPY INTO STARTER.SALES.RAW_SALES
FROM @nakamura_s3_stage
FILE_FORMAT = (
TYPE = 'CSV',
SKIP_HEADER = 1,
FIELD_OPTIONALLY_ENCLOSED_BY = '"'
)
ON_ERROR = 'CONTINUE';
--dbt Projectのtask
create or replace task <database>.<schema>."<dbt Project>"
warehouse=nakamura_wh
AFTER daily_copy_task
as EXECUTE dbt project <dbt Project> args='build --target dev';これでTASKの作成まで完了しました。実際に動くか確認しましょう。
このTASKを作成したデータベース>スキーマからルートタスクを選択してタスクグラフを実行すると強制実行ができます。
実行履歴で成功しましたと表示されていたら成功です。
また、dbtのTASKに関してはモニタリングのdbtプロジェクトから詳しい結果を確認できます。
dbtのTASKを使わない場合は？
dbtのTASKを使用しなくても、Snowflakeの通常のTASKでSQLやストアドプロシージャを用いて複数の処理を定義することで、ELT処理を構築することは可能です。
この方法ではTASKを多数作成せずに済む一方で、依存関係のある処理は個別に分ける必要があるほか、ストアドプロシージャを事前に定義する手間も発生します。
さらに、変換処理におけるテストの実行ができない点もデメリットです。
このような理由から、変換処理には dbtのTASKを活用する方が実用的かつ効率的 と言えるでしょう。
まとめ
今回はdbtを外部ステージのデータロードTASKと紐づけて実行してみました。
今回はこのような基本的な構成にしましたが、よりリアルタイム性を重視するならSnowpipeでロードをしたり、データが大きくて差分のみをロードすればよいならStreamオブジェクトを作成してテーブル更新をトリガーにdbtのTASKを実行したりするのも良いですね。
