Snowflakeで共有されたテーブルの変更を追跡して別テーブルを自動更新する

DA事業本部のnkhrです。Snowflakeではアカウント間でデータ共有ができます。本ブログでは、共有されたテーブルに対して変更追跡を行うためのストリームオブジェクトを設定し、変更履歴をもとに別テーブルを自動で更新する処理を検証しました。

Snowflakeでデータ共有する方法は3種類（Direct Share、Marketplace、Data Exchange）あります。本ブログで検証するデータ共有はDirect Shareです。

データ共有では、提供元をProviderアカウント、提供先をConsumerアカウントと呼びます。共有されたデータベースの実体はProviderアカウントにあります。Consumer側では、共有されたデータベース内にオブジェクトを作成したり、共有オブジェクトを修正することはできません。

今回の検証では、以下の図のイメージを実装します。

Providerアカウントの共有設定

共有したテーブルの変更履歴（STREAM）をConsumer側で追跡するためには共有するテーブルの「CHANGE_TRACKING」を「TRUE」にする必要があります。

共有設定の詳細は、以下のブログを参照してください。

Provider側では以下のオブジェクトを共有します。

• Database: test_db
• Schema: test
• Table: share
• Share object: test_s

実行コードは以下の通りです。

USE ROLE accountadmin;
GRANT CREATE SHARE ON ACCOUNT TO sysadmin;

USE ROLE SYSADMIN;
CREATE DATABASE test_db;
CREATE SCHEMA test;

CREATE table test_db.test.share(id INTEGER, name VARCHAR(20), class_number INTEGER);
INSERT INTO test.share VALUES (1, 'Juddy', 3), (2, 'Bob', 3), (3, 'Sam', 5);

ALTER TABLE test.SHARE SET CHANGE_TRACKING = TRUE;
CREATE OR REPLACE SHARE test_share;
GRANT USAGE ON DATABASE test_db TO SHARE test_share;
GRANT USAGE ON SCHEMA test_db.test TO SHARE test_share;
GRANT SELECT ON TABLE test_db.test.share TO SHARE test_share;
ALTER SHARE test_share SET ACCOUNTS = <sonsumer_accounts>;

Consumer側アカウントの設定

Consumer側では以下の作業を行います。

• Providerアカウントから共有されたShare Objectからデータベースを作成
• 共有テーブルに対してSTREAMを作成
• STREAMの変更をもとにConsumer側のテーブルを更新するTASKを作成

ストリームとタスクを利用したデータパイプラインの詳細については、下記のブログが参考になります。

ProviderアカウントのShare Objectからデータベース作成

Consumer側で共有データベースを参照するためには「IMPORTED PRIVILEGES」権限が必要です。共有データベースの利用権限はProvider側でShare Objectに設定されています。

USE ROLE ACCOUNTADMIN;
CREATE DATABASE shared_db FROM SHARE <provider account>.test_share;
GRANT IMPORTED PRIVILEGES ON DATABASE shared_db TO ROLE sysadmin;

共有データベースのテーブルに対してSTRAMを作成

共有されたデータベース内に、Consumer側でオブジェクトを作成できないため、STREAMはConsumer所有のデータベースに作成します。

Consumer側では、ユーザ一覧とユーザの変更日付を保持するuserlistテーブルを作成します。

USE ROLE SYSADMIN;
CREATE DATABASE c_db;
CREATE SCHEMA c_db.s_test;

CREATE TABLE c_db.s_test.userlist(id, name, updated_at)
  AS SELECT id, name current_timestamp(0) FROM shared_db.test.share;
CREATE STREAM c_db.s_test.mystream ON TABLE shared_db.test.share;

変更履歴からテーブルを更新するタスク作成

2021/12時点では、タスクの実行方法はスケジュール実行のみが提供されています。タスクの実行中に次のスケジュールの実行時間となった場合は、次の実行はスキップされます。TASK作成のWHEN句に「SYSTEM$STREAM_HAS_DATA(<stream name>)」を指定することで、対象のストリーム内にレコードが存在しない場合はタスク実行をスキップしてくれます。

スケジュールの指定は「X分ごとの実行」または「Cron式による実行」が指定できます。CREATE TASKの指定パラメータについてはリンク先をご参照ください。

2021/12時点では、アカウント内でStarted状態（実行中）のタスクは上限10000です（Suspend状態の場合はこのカウントに含まれません）

CREATE OR REPLACE TASK c_db.s_test.update_userlist
WAREHOUSE = TEST_WH
SCHEDULE = '1 minute'
ALLOW_OVERLAPPING_EXECUTION = FALSE
USER_TASK_TIMEOUT_MS = 600000
WHEN
SYSTEM$STREAM_HAS_DATA('c_db.s_test.mystream')
AS
MERGE INTO c_db.s_test.userlist u
// ストリームに同じid, nameのレコードが2レコード存在するのは
// class_numberカラムのみをUpdateした場合のため利用しないレコードとみなす
USING (
SELECT id, name, METADATA$ACTION AS operation_type
FROM c_db.s_test.mystream s1
WHERE NOT EXISTS (
    SELECT 1
    FROM c_db.s_test.mystream s2
    WHERE s1.id = s2.id AND s1.name = s2.name
    GROUP BY id, name
    HAVING COUNT(*) > 1
)) s
ON u.id = s.id AND u.name = s.name
WHEN MATCHED AND s.operation_type = 'DELETE' THEN DELETE
WHEN NOT MATCHED AND s.operation_type = 'INSERT'
THEN INSERT (id, name, updated_at) VALUES (s.id, s.name,  current_timestamp(0))
;

ALTER TASK IF EXISTS c_db.s_test.update_userlist RESUME;

動作確認

• Provider側での実行(2分ごとに各行を実行）

// ①追加
INSERT INTO test.share VALUES (4, 'Ken', 3), (5, 'Tonny', 5);

// ②削除
DELETE FROM test.share WHERE id = 1;

// ③更新
UPDATE test.share SET id = 1 WHERE id = 5;

// ④classnumberカラムの更新
UPDATE test.share SET class_number = 10 WHERE id = 2;

• consumer側での実行（Provider側で実行後に1分経過以降）

SELECT * FROM c_db.s_test.userlist;

①の結果（既存3行＋追加2行＝5行）

②の結果（1行削除＝＞4行、id = 2, 3, 4, 5）

③の結果（更新⇒4行、id = 1, 2, 3, 4）

④の結果（変化なし、id = 1, 2, 3, 4）

まとめ

共有されたテーブルの変更履歴から自動で、Consumer側のテーブルを更新する方法を検証しました。SnowflakeではOrganizationやReader Account機能により、複数アカウントの作成が簡単にできるため、アカウント間でのデータ共有をうまく利用することで、データの信頼性や処理効率化が図れるようになったら良いですね。

以上、@nkhrでした。