Snowflake の Stage でウォーターマークをチェックポイントとして管理する方法

概要

本記事では、Snowflake において、クエリベース CDC などの増分データ連携で使用するチェックポイントを、Stage 上の JSON ファイルで管理する方法を検証します。

増分データ連携では、前回どこまでデータを処理したかを把握するために、更新日時などの値をウォーターマーク（Watermark）として管理する必要があります。

本記事では、ソーステーブルごとにウォーターマークを含むチェックポイントを JSON ファイルとして保存し、次のライフサイクルを管理する方法を紹介します。

1. 初期状態の判定
2. チェックポイントの作成
3. チェックポイントの参照
4. チェックポイントの更新

事前準備

データベースとスキーマを作成

CREATE OR REPLACE DATABASE CDC_CHECKPOINT_LIFECYCLE_DB;
CREATE OR REPLACE SCHEMA CDC_CHECKPOINT_LIFECYCLE_DB.TEST_SCHEMA;

USE DATABASE CDC_CHECKPOINT_LIFECYCLE_DB;
USE SCHEMA TEST_SCHEMA;

Stage を作成

チェックポイントを JSON ファイルとして保存するための内部 Stage を作成します。FILE_FORMATを指定することで、クエリ実行時にファイルフォーマットを指定せずにクエリできます。

CREATE OR REPLACE STAGE CHECKPOINT_STAGE
    FILE_FORMAT = (TYPE = JSON COMPRESSION = NONE);

手順

初期状態の判定

最初に、チェックポイントファイルの存在を確認し、初回実行かどうかを判定します。取得結果が 0 件の場合はチェックポイントファイルが存在しないため、初期状態と判定します。この場合、前回の処理位置を示すウォーターマークが存在しないため、初回処理用の開始位置を設定する必要があります。

LIST @CHECKPOINT_STAGE/CDC_CHECKPOINT_LIFECYCLE_DB.TEST_SCHEMA.ORDERS/ 
    PATTERN = '.*checkpoint[.]json$';

SELECT COUNT(*)
FROM TABLE(RESULT_SCAN(LAST_QUERY_ID()));

チェックポイントの作成

チェックポイント情報を JSON 形式で Stage に出力します。

EXECUTE IMMEDIATE $$
DECLARE
    v_checkpoint VARIANT;
BEGIN
    -- チェックポイントの JSON を変数に格納
    SELECT OBJECT_CONSTRUCT_KEEP_NULL(
        'schema_version', 1,
        'source_table', 'UNLOAD_QUERY_CDC_TEST_DB.TEST_SCHEMA.ORDERS',
        'wartermark_column', 'AUDIT__UPDATA1_TIMESTAMP',
        'wartermark_value', '2026-10-04T17:00:00.000000000Z',
        'unloaded_rows', 3
    )
    INTO :v_checkpoint;

    -- JSON ファイルに出力
    COPY INTO
        @CHECKPOINT_STAGE/CDC_CHECKPOINT_LIFECYCLE_DB.TEST_SCHEMA.ORDERS/checkpoint.json
    FROM (
        SELECT :v_checkpoint
    )
    FILE_FORMAT = (TYPE = JSON COMPRESSION = NONE)
    SINGLE = TRUE
    OVERWRITE = TRUE;

    RETURN 'Checkpoint written successfully';
END;
$$;

チェックポイントの参照

作成したチェックポイントを参照し、保存された情報を確認します。

LIST @CHECKPOINT_STAGE/CDC_CHECKPOINT_LIFECYCLE_DB.TEST_SCHEMA.ORDERS/ 
    PATTERN = '.*checkpoint[.]json$';

name	size	md5	last_modified
checkpoint_stage/CDC_CHECKPOINT_LIFECYCLE_DB.TEST_SCHEMA.ORDERS/checkpoint.json	208	0e3bfff8cc0a731777b0e067b5698ec3	Sat, 10 Oct 2026 07:49:13 GMT

SELECT
    $1 AS value,
    $1:schema_version::INT AS schema_version,
    $1:source_table::STRING AS source_table,
    $1:wartermark_column::STRING AS wartermark_column,
    $1:wartermark_value::STRING AS wartermark_value,
    $1:unloaded_rows::INT AS unloaded_rows
FROM
    @CHECKPOINT_STAGE/CDC_CHECKPOINT_LIFECYCLE_DB.TEST_SCHEMA.ORDERS/checkpoint.json;

VALUE	SCHEMA_VERSION	SOURCE_TABLE	WARTERMARK_COLUMN	WARTERMARK_VALUE	UNLOADED_ROWS
{ "schema_version": 1, "source_table": "UNLOAD_QUERY_CDC_TEST_DB.TEST_SCHEMA.ORDERS", "unloaded_rows": 3, "wartermark_column": "AUDIT__UPDATA1_TIMESTAMP", "wartermark_value": "2026-10-04T17:00:00.000000000Z" }	1	UNLOAD_QUERY_CDC_TEST_DB.TEST_SCHEMA.ORDERS	AUDIT__UPDATA1_TIMESTAMP	2026-10-04T17:00:00.000000000Z	3

ウォーターマークの更新

増分データ連携の処理が完了したことを想定し、チェックポイントに保存されているウォーターマークを更新します。OVERWRITE = TRUE を指定しているため、同名のチェックポイントファイルが存在する場合は、新しいチェックポイント情報で上書きされます。

EXECUTE IMMEDIATE $$
DECLARE
    v_checkpoint VARIANT;
BEGIN
    -- チェックポイントの JSON を変数に格納
    SELECT OBJECT_CONSTRUCT_KEEP_NULL(
        'schema_version', 1,
        'source_table', 'UNLOAD_QUERY_CDC_TEST_DB.TEST_SCHEMA.ORDERS',
        'wartermark_column', 'AUDIT__UPDATA1_TIMESTAMP',
        'value', '2026-10-10T17:00:00.000000000Z',
        'unloaded_rows', 100
    )
    INTO :v_checkpoint;

    -- JSON ファイルに出力
    COPY INTO
        @CHECKPOINT_STAGE/CDC_CHECKPOINT_LIFECYCLE_DB.TEST_SCHEMA.ORDERS/checkpoint.json
    FROM (
        SELECT :v_checkpoint
    )
    FILE_FORMAT = (TYPE = JSON COMPRESSION = NONE)
    SINGLE = TRUE
    OVERWRITE = TRUE;

    RETURN 'Checkpoint written successfully';
END;
$$;

チェックポイントファイルが同じパスに存在し、ファイルサイズや最終更新日時などが変化したことを確認します。この仕組みにより、次回の増分データ連携では、更新されたウォーターマークを基準に処理を再開できます。

LIST @CHECKPOINT_STAGE/CDC_CHECKPOINT_LIFECYCLE_DB.TEST_SCHEMA.ORDERS/ 
    PATTERN = '.*checkpoint[.]json$';

name	size	md5	last_modified
checkpoint_stage/CDC_CHECKPOINT_LIFECYCLE_DB.TEST_SCHEMA.ORDERS/checkpoint.json	192	0c6b3fcb8659845f28be9d06f07aabb3	Sat, 10 Oct 2026 07:49:14 GMT

SELECT
    $1 AS value,
    $1:schema_version::INT AS schema_version,
    $1:source_table::STRING AS source_table,
    $1:wartermark_column::STRING AS wartermark_column,
    $1:wartermark_value::STRING AS wartermark_value,
    $1:unloaded_rows::INT AS unloaded_rows
FROM
    @CHECKPOINT_STAGE/CDC_CHECKPOINT_LIFECYCLE_DB.TEST_SCHEMA.ORDERS/checkpoint.json;

VALUE	SCHEMA_VERSION	SOURCE_TABLE	WARTERMARK_COLUMN	WARTERMARK_VALUE	UNLOADED_ROWS
{ "schema_version": 1, "source_table": "UNLOAD_QUERY_CDC_TEST_DB.TEST_SCHEMA.ORDERS", "unloaded_rows": 0, "value": "2026-10-10T17:00:00.000000000Z", "wartermark_column": "AUDIT__UPDATA1_TIMESTAMP" }	1	UNLOAD_QUERY_CDC_TEST_DB.TEST_SCHEMA.ORDERS	AUDIT__UPDATA1_TIMESTAMP		0