ステージ、ファイル形式、COPY INTO

Snowflakeにおけるデータパイプラインの自動化

Emily Melhuish

Technical Curriculum Developer, Snowflake

Snowflakeが支える多様なワークロード

Screenshot 2026-05-11 at 11.52.34 am.png

1 * Snowflake Learning Material
Snowflakeにおけるデータパイプラインの自動化

パイプラインの例

 

 

Screenshot 2026-05-11 at 10.48.51 am.png

1 * Snowflake Learning Material
Snowflakeにおけるデータパイプラインの自動化

Harbrのご紹介

Harbrの架空ロゴ

Harbrについて

  • グローバルな物流・サプライチェーンプラットフォーム
  • Snowflakeを活用して出荷、倉庫在庫、配送パフォーマンスを管理
  • 複数地域のサプライヤーシステムからデータを取り込む
Snowflakeにおけるデータパイプラインの自動化

データ読み込みの課題

データ読み込みの図

ステージ = 一時的なストレージの場所

Snowflakeにおけるデータパイプラインの自動化

ステージの種類

2種類のステージ

内部ステージ

  • ユーザー
    • 単一ユーザー用の個人ステージング領域
  • テーブル
    • 特定のテーブルに紐付く
  • 名前付き
    • スキーマ内に作成されるデータベースオブジェクト

外部ステージ

  • 名前付き
    • クラウドプロバイダーを参照
Snowflakeにおけるデータパイプラインの自動化

ステージのパラメータ

ディレクトリテーブル

CREATE STAGE harbr_stage
  DIRECTORY = (ENABLE = TRUE)

内部ステージの暗号化

ENCRYPTION = (TYPE = SNOWFLAKE_FULL)
ENCRYPTION = (TYPE = SNOWFLAKE_SSE)

外部ステージの暗号化

ENCRYPTION = 
([ TYPE = 'AWS_CSE' ] MASTER_KEY = '') 

ENCRYPTION = 
([ TYPE = 'AWS_SSE_S3' ]) 

...
Snowflakeにおけるデータパイプラインの自動化

ファイル形式

  • 構造化形式：区切り文字・ヘッダー行・引用ルールを指定したCSV

  • 半構造化形式：JSON、Parquet、Avro、ORC、XML

    • VARIANT 列に直接読み込み可能
    • 解析ルールを一度定義し、すべての読み込み操作で再利用

  • フォーマットオブジェクトを一度更新するだけで反映

SQL

CREATE FILE FORMAT harbr_csv_format
  TYPE = 'CSV'
  FIELD_DELIMITER = ','
  SKIP_HEADER = 1;
Snowflakeにおけるデータパイプラインの自動化

COPY INTO

名前付きステージと名前付きファイル形式を使った読み込み

COPY INTO logistics.shipments
FROM @harbr_internal_stage/shipments/
FILE_FORMAT = (FORMAT_NAME = 'harbr_csv_format')
Snowflakeにおけるデータパイプラインの自動化

エラー処理

ON_ERROR オプション 動作
ABORT_STATEMENT 最初のエラーで読み込み全体を中止 (デフォルト)
CONTINUE 不正な行をスキップし、残りを読み込む
SKIP_FILE エラーを含むファイル全体をスキップ
SKIP_FILE_<num> エラー数が指定件数を超えた場合のみスキップ
SKIP_FILE_<num>% エラー率が指定の閾値を超えた場合のみスキップ
Snowflakeにおけるデータパイプラインの自動化

検証モードとCOPY_HISTORY

読み込まずに検証 = ドライラン

COPY INTO logistics.shipments
FROM @harbr_internal_stage/shipments/
FILE_FORMAT = (FORMAT_NAME = 'harbr_csv_format')
VALIDATION_MODE = RETURN_ERRORS;

読み込み履歴の確認

SELECT * FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
  TABLE_NAME => 'shipments',
  START_TIME => DATEADD('hour', -24,
    CURRENT_TIMESTAMP())));
Snowflakeにおけるデータパイプラインの自動化

練習しましょう！

Snowflakeにおけるデータパイプラインの自動化

Preparing Video For Download...