データアンロードと接続性

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

Emily Melhuish

Technical Curriculum Developer, Snowflake

アンロードのユースケース

クラウドストレージへのデータエクスポート

Snowflakeにアクセスできないパートナーも存在する
COPY INTO で結果をステージに直接書き出す

別途エクスポートパイプライン不要

COPY INTO @harbr_partner_export/
daily_summary/
FROM (SELECT * FROM 
logistics.shipment_summary
WHERE export_date = 
CURRENT_DATE() - 1);

mermaid: パートナーストレージへのSnowflakeアンロード

COPY INTO

COPY INTO @harbr_partner_export/shipment_summary/
FROM (
  SELECT shipment_id,
         origin,
         destination,
         delivery_status,
         delivery_time_hours
  FROM logistics.shipments
  WHERE delivery_date = CURRENT_DATE()
)
FILE_FORMAT = (TYPE = 'CSV' HEADER = TRUE)
OVERWRITE = TRUE;

エクスポートファイル形式のオプション

形式	最適な用途
CSV	汎用 — ほぼすべてのシステムで読み込み可能；パートナーエクスポートに最適
JSON	ネスト構造を保持；半構造化データの出力に有効
Parquet	大規模分析データセット；列指向圧縮でファイルサイズ削減・高速読み込み

主なアンロードオプション

-- Split large exports across multiple files (bytes)
HEADER = TRUE              -- include column names in first row
OVERWRITE = TRUE           -- replace any existing files at that path
MAX_FILE_SIZE = 104857600  -- 100 MB per output file

接続性の全体像

接続性の全体像：コネクタ

KafkaおよびSparkコネクタ

Kafkaコネクタ

トピックをSnowflakeテーブルに直接ストリーミング
ファイル・ステージ不要 — Snowpipe Streamingを使用
遅延は数秒単位

# connector.properties (Kafka settings)
snowflake.topic2table.map=events:
delivery_events
snowflake.ingestion.method=
SNOWPIPE_STREAMING

Sparkコネクタ

SparkのDataFrame APIと統合
SparkジョブからSnowflakeへの読み書きが可能
大規模な変換ワークロードに対応

// Read from Snowflake into a Spark DataFrame
val df = spark.read.format("snowflake")
  .options(sfOptions).option("dbtable", 
    "shipments").load()

JDBC/ODBCとパートナー統合

ユニバーサル接続

統合	種類	Harbr での用途
JDBC / ODBC	汎用ドライバー	BIツール（Tableau、Power BI、Looker）— 直接クエリ
Pythonコネクタ	ネイティブPythonドライバー	データパイプライン、定期ETL、データサイエンス
dbt	SQL変換	Snowflakeコンピュート上でモデルを直接実行
Fivetran / Airbyte	マネージドインジェスト	SaaSソース → Snowflake、カスタムコード不要

練習しましょう！

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