Snowflake の STREAM を使って CDC データを Parquet ファイルへ UNLOAD する方法

概要

Snowflake のテーブルを対象に変更データキャプチャ（CDC）を行うため、STREAM をソースとして Parquet ファイルへ UNLOAD する方法を紹介します。

事前準備

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

検証用のデータベースとスキーマを作成します。

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

USE DATABASE UNLOAD_TEST_DB;
USE SCHEMA TEST_SCHEMA;

テーブルと STREAM を作成

変更追跡（CHANGE_TRACKING）を有効にしたテーブルと、そのテーブルを対象とする STREAM を作成します。

CREATE OR REPLACE TABLE ORDERS (
    ORDER_ID                 NUMBER,
    CUSTOMER_NAME            VARCHAR,
    AMOUNT                   NUMBER(18, 2),
    ORDER_DATE               DATE,
    AUDIT__INGEST_TIMESTAMP  TIMESTAMP_NTZ,
    AUDIT__UPDATE_TIMESTAMP  TIMESTAMP_NTZ
)
CHANGE_TRACKING = TRUE;

CREATE OR REPLACE STREAM ORDERS_STREAM
ON TABLE ORDERS
APPEND_ONLY = FALSE;

UNLOAD 用のストアドプロシージャを定義

STREAM の差分データを Parquet ファイルとしてステージへ出力するストアドプロシージャを定義します。出力先のディレクトリは、実行時刻（UTC）でパーティション分割します。

CREATE OR REPLACE PROCEDURE SP_UNLOAD(
    P_TARGET_STAGE         VARCHAR,
    P_TARGET_PATH          VARCHAR,
    P_SOURCE_QUERY         VARCHAR,
    P_FILE_FORMAT          VARCHAR DEFAULT NULL,
    P_PARTITION_BY         VARCHAR DEFAULT NULL,
    P_OVERWRITE            BOOLEAN DEFAULT NULL,
    P_SINGLE               BOOLEAN DEFAULT NULL,
    P_MAX_FILE_SIZE        NUMBER(38, 0) DEFAULT NULL,
    P_INCLUDE_QUERY_ID     BOOLEAN DEFAULT NULL,
    P_DETAILED_OUTPUT      BOOLEAN DEFAULT NULL,
    P_VALIDATION_MODE      VARCHAR DEFAULT NULL,
    P_HEADER               BOOLEAN DEFAULT NULL,
    P_TARGET_LOCATION      VARCHAR DEFAULT NULL,
    P_STORAGE_INTEGRATION  VARCHAR DEFAULT NULL,
    P_CREDENTIALS          VARCHAR DEFAULT NULL,
    P_ENCRYPTION           VARCHAR DEFAULT NULL
)
RETURNS VARCHAR
LANGUAGE SQL
EXECUTE AS OWNER
AS
$$
DECLARE
    copy_sql        VARCHAR;
    target_location VARCHAR;
    copy_query_id   VARCHAR;

BEGIN
    -- =========================================================
    -- 1. 出力先を設定
    --
    -- ステージへの出力:
    --   P_TARGET_STAGE と P_TARGET_PATH を使用
    --
    -- 外部 URL への直接出力:
    --   P_TARGET_LOCATION を使用
    -- =========================================================
    IF (P_TARGET_LOCATION IS NOT NULL) THEN
        target_location := P_TARGET_LOCATION;
    ELSE
        target_location :=
            '@' || P_TARGET_STAGE || '/' ||
            LTRIM(COALESCE(P_TARGET_PATH, ''), '/');
    END IF;

    copy_sql :=
        'COPY INTO ' ||
        CHR(39) ||
        REPLACE(
            target_location,
            CHR(39),
            CHR(39) || CHR(39)
        ) ||
        CHR(39);

    -- =========================================================
    -- 2. 外部 URL 用の認証・暗号化設定
    -- =========================================================
    IF (P_STORAGE_INTEGRATION IS NOT NULL) THEN
        copy_sql := copy_sql ||
            ' STORAGE_INTEGRATION = ' || P_STORAGE_INTEGRATION;
    END IF;

    IF (P_CREDENTIALS IS NOT NULL) THEN
        copy_sql := copy_sql ||
            ' CREDENTIALS = (' || P_CREDENTIALS || ')';
    END IF;

    IF (P_ENCRYPTION IS NOT NULL) THEN
        copy_sql := copy_sql ||
            ' ENCRYPTION = (' || P_ENCRYPTION || ')';
    END IF;

    -- =========================================================
    -- 3. SELECT 文・パーティション式・ファイル形式
    -- =========================================================
    copy_sql := copy_sql ||
        ' FROM (' || P_SOURCE_QUERY || ')';

    IF (P_PARTITION_BY IS NOT NULL) THEN
        copy_sql := copy_sql ||
            ' PARTITION BY (' || P_PARTITION_BY || ')';
    END IF;

    IF (P_FILE_FORMAT IS NOT NULL) THEN
        copy_sql := copy_sql ||
            ' FILE_FORMAT = (' || P_FILE_FORMAT || ')';
    END IF;

    -- =========================================================
    -- 4. COPY オプション
    -- =========================================================
    IF (P_OVERWRITE IS NOT NULL) THEN
        copy_sql := copy_sql ||
            ' OVERWRITE = ' ||
            IFF(P_OVERWRITE, 'TRUE', 'FALSE');
    END IF;

    IF (P_SINGLE IS NOT NULL) THEN
        copy_sql := copy_sql ||
            ' SINGLE = ' ||
            IFF(P_SINGLE, 'TRUE', 'FALSE');
    END IF;

    IF (P_MAX_FILE_SIZE IS NOT NULL) THEN
        copy_sql := copy_sql ||
            ' MAX_FILE_SIZE = ' ||
            TO_VARCHAR(P_MAX_FILE_SIZE);
    END IF;

    IF (P_INCLUDE_QUERY_ID IS NOT NULL) THEN
        copy_sql := copy_sql ||
            ' INCLUDE_QUERY_ID = ' ||
            IFF(P_INCLUDE_QUERY_ID, 'TRUE', 'FALSE');
    END IF;

    IF (P_DETAILED_OUTPUT IS NOT NULL) THEN
        copy_sql := copy_sql ||
            ' DETAILED_OUTPUT = ' ||
            IFF(P_DETAILED_OUTPUT, 'TRUE', 'FALSE');
    END IF;

    -- =========================================================
    -- 5. 検証モード・列名出力
    -- =========================================================
    IF (P_VALIDATION_MODE IS NOT NULL) THEN
        copy_sql := copy_sql ||
            ' VALIDATION_MODE = ' ||
            CHR(39) ||
            REPLACE(
                P_VALIDATION_MODE,
                CHR(39),
                CHR(39) || CHR(39)
            ) ||
            CHR(39);
    END IF;

    IF (P_HEADER IS NOT NULL) THEN
        copy_sql := copy_sql ||
            ' HEADER = ' ||
            IFF(P_HEADER, 'TRUE', 'FALSE');
    END IF;

    -- =========================================================
    -- 6. COPY INTO を実行し、クエリ ID を返す
    -- =========================================================
    EXECUTE IMMEDIATE :copy_sql;

    copy_query_id := SQLID;

    RETURN copy_query_id;

END;
$$;

動作確認

1. INSERT の実施

データを挿入します。

INSERT INTO ORDERS (
    ORDER_ID,
    CUSTOMER_NAME,
    AMOUNT,
    ORDER_DATE,
    AUDIT__INGEST_TIMESTAMP,
    AUDIT__UPDATE_TIMESTAMP
)
VALUES
    (
        1,
        'Alice',
        100.50,
        '2026-10-01',
        '2026-10-04 17:00:00'::TIMESTAMP_LTZ,
        '2026-10-04 17:00:00'::TIMESTAMP_LTZ
    ),
    (
        2,
        'Bob',
        200.00,
        '2026-10-02',
        '2026-10-04 17:00:00'::TIMESTAMP_LTZ,
        '2026-10-04 17:00:00'::TIMESTAMP_LTZ
    ),
    (
        3,
        'Carol',
        300.00,
        '2026-10-03',
        '2026-10-04 17:00:00'::TIMESTAMP_LTZ,
        '2026-10-04 17:00:00'::TIMESTAMP_LTZ
    );

STREAM で差分データを確認します。

SELECT * FROM ORDERS_STREAM;

ストアドプロシージャを実行します。

CALL SP_UNLOAD(
    P_TARGET_STAGE => 'UNLOAD_TEST_DB.TEST_SCHEMA.ORDERS_CDC_STAGE',
    P_TARGET_PATH  => 'output/',

    P_SOURCE_QUERY => $$
        SELECT
            ORDER_ID,
            CUSTOMER_NAME,
            AMOUNT,
            ORDER_DATE,
            CONVERT_TIMEZONE(
                'UTC',
                AUDIT__INGEST_TIMESTAMP
            )::TIMESTAMP_NTZ(0) AS AUDIT__INGEST_TIMESTAMP,
            CONVERT_TIMEZONE(
                'UTC',
                AUDIT__UPDATE_TIMESTAMP
            )::TIMESTAMP_NTZ(0) AS AUDIT__UPDATE_TIMESTAMP,
            METADATA$ACTION AS AUDIT_METADATA_ACTION,
            METADATA$ISUPDATE AS AUDIT_METADATA_ISUPDATE,
            METADATA$ROW_ID AS AUDIT_METADATA_ROW_ID
        FROM
            UNLOAD_TEST_DB.TEST_SCHEMA.ORDERS_STREAM
    $$,

    P_FILE_FORMAT => $$
        TYPE = PARQUET
        COMPRESSION = SNAPPY
    $$,

    P_PARTITION_BY => $$
        'audit__ingest_timestamp=' ||
        TO_VARCHAR(
            CONVERT_TIMEZONE('UTC', CURRENT_TIMESTAMP()),
            'YYYY-MM-DD"T"HH24:MI:SSZ'
        )
    $$,
    P_MAX_FILE_SIZE       => 268435456,
    P_INCLUDE_QUERY_ID    => TRUE
);

ステージにファイルが追加されたことを確認します。

LIST @ORDERS_CDC_STAGE/output

追加されたファイルのデータを表示します。

CREATE OR REPLACE FILE FORMAT PARQUET_READ_FORMAT TYPE = PARQUET;
SELECT
    T.$1
FROM
    @orders_cdc_stage/output/audit__ingest_timestamp=2026-10-04T08:50:25Z/  -- ToDo 追加されたディレクトリに変更
    (
        FILE_FORMAT => 'PARQUET_READ_FORMAT'
    ) AS T;

{
  "AMOUNT": 100.5,
  "AUDIT_METADATA_ACTION": "INSERT",
  "AUDIT_METADATA_ISUPDATE": false,
  "AUDIT_METADATA_ROW_ID": "c4eb907dd170c30562daa215ab5ee600e11d7406",
  "AUDIT__INGEST_TIMESTAMP": "2026-10-05 00:00:00.000",
  "AUDIT__UPDATE_TIMESTAMP": "2026-10-05 00:00:00.000",
  "CUSTOMER_NAME": "Alice",
  "ORDER_DATE": "2026-10-01",
  "ORDER_ID": 1
}
{
  "AMOUNT": 200,
  "AUDIT_METADATA_ACTION": "INSERT",
  "AUDIT_METADATA_ISUPDATE": false,
  "AUDIT_METADATA_ROW_ID": "ea2ddf0554589d1bca34a5b5cda2f6d333affe7b",
  "AUDIT__INGEST_TIMESTAMP": "2026-10-05 00:00:00.000",
  "AUDIT__UPDATE_TIMESTAMP": "2026-10-05 00:00:00.000",
  "CUSTOMER_NAME": "Bob",
  "ORDER_DATE": "2026-10-02",
  "ORDER_ID": 2
}
{
  "AMOUNT": 300,
  "AUDIT_METADATA_ACTION": "INSERT",
  "AUDIT_METADATA_ISUPDATE": false,
  "AUDIT_METADATA_ROW_ID": "f6627c0bfa5171498f76d21f7ba598166ad2f601",
  "AUDIT__INGEST_TIMESTAMP": "2026-10-05 00:00:00.000",
  "AUDIT__UPDATE_TIMESTAMP": "2026-10-05 00:00:00.000",
  "CUSTOMER_NAME": "Carol",
  "ORDER_DATE": "2026-10-03",
  "ORDER_ID": 3
}

2. UPSERT の実施

MERGE 文で UPSERT（既存レコードの更新と新規レコードの挿入）を実施します。

MERGE INTO ORDERS AS T
USING (
    SELECT
        COLUMN1::NUMBER             AS ORDER_ID,
        COLUMN2::VARCHAR            AS CUSTOMER_NAME,
        COLUMN3::NUMBER(18, 2)      AS AMOUNT,
        COLUMN4::DATE               AS ORDER_DATE,
        COLUMN5::TIMESTAMP_LTZ      AS AUDIT__INGEST_TIMESTAMP,
        COLUMN6::TIMESTAMP_LTZ      AS AUDIT__UPDATE_TIMESTAMP
    FROM VALUES
        (
            '2',
            'Bob Updated',
            '250.00',
            '2026-10-02',
            '2026-10-04 17:10:00',
            '2026-10-04 17:10:00'
        ),
        (
            '4',
            'Dave',
            '400.00',
            '2026-10-04',
            '2026-10-04 17:10:00',
            '2026-10-04 17:10:00'
        )
) AS S
    ON T.ORDER_ID = S.ORDER_ID

WHEN MATCHED THEN
    UPDATE SET
        T.CUSTOMER_NAME           = S.CUSTOMER_NAME,
        T.AMOUNT                  = S.AMOUNT,
        T.ORDER_DATE              = S.ORDER_DATE,
        T.AUDIT__INGEST_TIMESTAMP = S.AUDIT__INGEST_TIMESTAMP,
        T.AUDIT__UPDATE_TIMESTAMP = S.AUDIT__UPDATE_TIMESTAMP

WHEN NOT MATCHED THEN
    INSERT (
        ORDER_ID,
        CUSTOMER_NAME,
        AMOUNT,
        ORDER_DATE,
        AUDIT__INGEST_TIMESTAMP,
        AUDIT__UPDATE_TIMESTAMP
    )
    VALUES (
        S.ORDER_ID,
        S.CUSTOMER_NAME,
        S.AMOUNT,
        S.ORDER_DATE,
        S.AUDIT__INGEST_TIMESTAMP,
        S.AUDIT__UPDATE_TIMESTAMP
    );

STREAM で差分データを確認します。

SELECT * FROM ORDERS_STREAM;

ストアドプロシージャを実行します。

CALL SP_UNLOAD(
    P_TARGET_STAGE => 'UNLOAD_TEST_DB.TEST_SCHEMA.ORDERS_CDC_STAGE',
    P_TARGET_PATH  => 'output/',

    P_SOURCE_QUERY => $$
        SELECT
            ORDER_ID,
            CUSTOMER_NAME,
            AMOUNT,
            ORDER_DATE,
            CONVERT_TIMEZONE(
                'UTC',
                AUDIT__INGEST_TIMESTAMP
            )::TIMESTAMP_NTZ(0) AS AUDIT__INGEST_TIMESTAMP,
            CONVERT_TIMEZONE(
                'UTC',
                AUDIT__UPDATE_TIMESTAMP
            )::TIMESTAMP_NTZ(0) AS AUDIT__UPDATE_TIMESTAMP,
            METADATA$ACTION AS AUDIT_METADATA_ACTION,
            METADATA$ISUPDATE AS AUDIT_METADATA_ISUPDATE,
            METADATA$ROW_ID AS AUDIT_METADATA_ROW_ID
        FROM
            UNLOAD_TEST_DB.TEST_SCHEMA.ORDERS_STREAM
    $$,

    P_FILE_FORMAT => $$
        TYPE = PARQUET
        COMPRESSION = SNAPPY
    $$,

    P_PARTITION_BY => $$
        'audit__ingest_timestamp=' ||
        TO_VARCHAR(
            CONVERT_TIMEZONE('UTC', CURRENT_TIMESTAMP()),
            'YYYY-MM-DD"T"HH24:MI:SSZ'
        )
    $$,
    P_MAX_FILE_SIZE       => 268435456,
    P_INCLUDE_QUERY_ID    => TRUE
);

ステージにファイルが追加されたことを確認します。

LIST @ORDERS_CDC_STAGE/output

追加されたファイルのデータを表示します。更新されたレコード（ORDER_ID = 2）は、AUDIT_METADATA_ISUPDATE が true の DELETE（更新前）と INSERT（更新後）の 2 行として出力されます。

CREATE OR REPLACE FILE FORMAT PARQUET_READ_FORMAT TYPE = PARQUET;
SELECT
    T.$1
FROM
    @orders_cdc_stage/output/audit__ingest_timestamp=2026-10-04T09:04:52Z/  -- ToDo 追加されたディレクトリに変更
    (
        FILE_FORMAT => 'PARQUET_READ_FORMAT'
    ) AS T;

{
  "AMOUNT": 400,
  "AUDIT_METADATA_ACTION": "INSERT",
  "AUDIT_METADATA_ISUPDATE": false,
  "AUDIT_METADATA_ROW_ID": "b922d9531767779c0df44dcd3468cc3dd0ddf89d",
  "AUDIT__INGEST_TIMESTAMP": "2026-10-05 00:10:00.000",
  "AUDIT__UPDATE_TIMESTAMP": "2026-10-05 00:10:00.000",
  "CUSTOMER_NAME": "Dave",
  "ORDER_DATE": "2026-10-04",
  "ORDER_ID": 4
}
{
  "AMOUNT": 250,
  "AUDIT_METADATA_ACTION": "INSERT",
  "AUDIT_METADATA_ISUPDATE": true,
  "AUDIT_METADATA_ROW_ID": "3226536b5c635cd0cdc2d11a3465796c0aad047a",
  "AUDIT__INGEST_TIMESTAMP": "2026-10-05 00:10:00.000",
  "AUDIT__UPDATE_TIMESTAMP": "2026-10-05 00:10:00.000",
  "CUSTOMER_NAME": "Bob Updated",
  "ORDER_DATE": "2026-10-02",
  "ORDER_ID": 2
}
{
  "AMOUNT": 200,
  "AUDIT_METADATA_ACTION": "DELETE",
  "AUDIT_METADATA_ISUPDATE": true,
  "AUDIT_METADATA_ROW_ID": "3226536b5c635cd0cdc2d11a3465796c0aad047a",
  "AUDIT__INGEST_TIMESTAMP": "2026-10-05 00:00:00.000",
  "AUDIT__UPDATE_TIMESTAMP": "2026-10-05 00:00:00.000",
  "CUSTOMER_NAME": "Bob",
  "ORDER_DATE": "2026-10-02",
  "ORDER_ID": 2
}

3. DELETE の実施

DELETE を実施します。

DELETE FROM ORDERS
WHERE ORDER_ID = 1;

STREAM で差分データを確認します。

SELECT * FROM ORDERS_STREAM;

ストアドプロシージャを実行します。

CALL SP_UNLOAD(
    P_TARGET_STAGE => 'UNLOAD_TEST_DB.TEST_SCHEMA.ORDERS_CDC_STAGE',
    P_TARGET_PATH  => 'output/',

    P_SOURCE_QUERY => $$
        SELECT
            ORDER_ID,
            CUSTOMER_NAME,
            AMOUNT,
            ORDER_DATE,
            CONVERT_TIMEZONE(
                'UTC',
                AUDIT__INGEST_TIMESTAMP
            )::TIMESTAMP_NTZ(0) AS AUDIT__INGEST_TIMESTAMP,
            CONVERT_TIMEZONE(
                'UTC',
                AUDIT__UPDATE_TIMESTAMP
            )::TIMESTAMP_NTZ(0) AS AUDIT__UPDATE_TIMESTAMP,
            METADATA$ACTION AS AUDIT_METADATA_ACTION,
            METADATA$ISUPDATE AS AUDIT_METADATA_ISUPDATE,
            METADATA$ROW_ID AS AUDIT_METADATA_ROW_ID
        FROM
            UNLOAD_TEST_DB.TEST_SCHEMA.ORDERS_STREAM
    $$,

    P_FILE_FORMAT => $$
        TYPE = PARQUET
        COMPRESSION = SNAPPY
    $$,

    P_PARTITION_BY => $$
        'audit__ingest_timestamp=' ||
        TO_VARCHAR(
            CONVERT_TIMEZONE('UTC', CURRENT_TIMESTAMP()),
            'YYYY-MM-DD"T"HH24:MI:SSZ'
        )
    $$,
    P_MAX_FILE_SIZE       => 268435456,
    P_INCLUDE_QUERY_ID    => TRUE
);

ステージにファイルが追加されたことを確認します。

LIST @ORDERS_CDC_STAGE/output

追加されたファイルのデータを表示します。

CREATE OR REPLACE FILE FORMAT PARQUET_READ_FORMAT TYPE = PARQUET;
SELECT
    T.$1
FROM
    @orders_cdc_stage/output/audit__ingest_timestamp=2026-10-04T09:04:57Z/  -- ToDo 追加されたディレクトリに変更
    (
        FILE_FORMAT => 'PARQUET_READ_FORMAT'
    ) AS T;

{
  "AMOUNT": 100.5,
  "AUDIT_METADATA_ACTION": "DELETE",
  "AUDIT_METADATA_ISUPDATE": false,
  "AUDIT_METADATA_ROW_ID": "04e6f4b1b375731d9ec770be0b561a6066140ead",
  "AUDIT__INGEST_TIMESTAMP": "2026-10-05 00:00:00.000",
  "AUDIT__UPDATE_TIMESTAMP": "2026-10-05 00:00:00.000",
  "CUSTOMER_NAME": "Alice",
  "ORDER_DATE": "2026-10-01",
  "ORDER_ID": 1
}