SnowflakeへのAppFlowのデータ連携をトリガーにTASKを実行する
データ事業本部の鈴木です。
昨日SalesforceのデータをAmazon AppFlowでSnowflakeに増分連携してみた話を公開しましたが、連携後にTASK実行して追加の処理を実行したかったので、キックできるか確認しました。
▼昨日公開した記事
実現方法について
AppFlowとSnowflakeを組み合わせるとき、AppFlowからのデータ連携を起点にSnowflakeの処理を実行する方法は大きく2種あると思います。
- Snowflake側でストリームでテーブルの変更を検知し、タスクを実行する
- AWS側でフローの完了を検知し、AWSからSnowflakeの処理を実行する
後者は例えばEventBridgeを発火させてLambdaを実行するなどが考えられますが、AWSでSnowflakeの認証情報を持っていないとSnowflakeのAPIが実行できないため、前者に寄せた方が楽そうです。
SnowflakeではストリームでテーブルへのCOPY INTOを検知できるため、これを起点にタスクを実行できそうでした。
ストリームからタスクの実行は、以下を参考に試してみました。
やってみる
1. Snowflakeのリソースの準備
以下のようにSnowflake側でストリーム・タスクを作成しておきました。ストリームが監視するテーブルは別のブログで作成してあり、AppFlowフローでSalesforceからデータ連携しているとします。
-- ストリーム作成
CREATE OR REPLACE STREAM OPPORTUNITY_STREAM ON TABLE APPFLOW_DB
.APPFLOW_SCHEMA.SALESFORCE_OPPORTUNITY;
-- データ消費向けのテーブル
CREATE OR REPLACE TABLE APPFLOW_DB.APPFLOW_SCHEMA.APPFLOW_LOAD_EVENTS (
detected_at TIMESTAMP_NTZ,
stream_row_count NUMBER
);
-- タスク定義
CREATE OR REPLACE TASK APPFLOW_DB.APPFLOW_SCHEMA.CONSUME_OPPORTUNITY_STREAM
WAREHOUSE = COMPUTE_WH
WHEN SYSTEM$STREAM_HAS_DATA('APPFLOW_DB.APPFLOW_SCHEMA.OPPORTUNITY_STREAM')
AS
INSERT INTO APPFLOW_DB.APPFLOW_SCHEMA.APPFLOW_LOAD_EVENTS
SELECT
CURRENT_TIMESTAMP(),
COUNT(*)
FROM APPFLOW_DB.APPFLOW_SCHEMA.OPPORTUNITY_STREAM;
-- タスクのRESUME
ALTER TASK CONSUME_OPPORTUNITY_STREAM RESUME;
2. Snowflakeへのデータ連携
Salesforceで検証向けにデータを作成、AppFlowでSnowflakeへ連携しました。
▼作成したデータ例
▼AppFlowでの連携履歴
3. タスクの実行確認
タスクの実行履歴を確認すると、確かにINSERTが実行されていました。
▼タスクの実行履歴
▼実行されたタスクログ
最後に
ストリームを使い、SnowflakeへのAppFlowのデータ連携をトリガーにTASKを実行してみました。
今回は単にデータをSELECTしただけでしたが、このタスクの後にdbtプロジェクトの実行のタスクを繋げればdbt実行のトリガーにもできるので、パイプライン構築の幅が広がりそうです。
参考になりましたら幸いです。
