Snowflake

Same Session as getting started. Add a sink. Skipprd stages Parquet and runs COPY INTO. Schema DDL runs before the data.

Credentials

export SNOWFLAKE_ACCOUNT="myorg-myaccount"
export SNOWFLAKE_USER="skippr_loader"
export SNOWFLAKE_PRIVATE_KEY_PATH="/path/to/rsa_key.p8"
export AWS_ACCESS_KEY_ID="your-key"
export AWS_SECRET_ACCESS_KEY="your-secret"
export AWS_DEFAULT_REGION="us-east-1"

Key-pair auth. Flags: Snowflake quickstart.

Write skippr.yml

Python

from skippr import Config, DataSinkSnowflake, DataSourceS3, Pipeline, S3Storage

cfg = Config().workspace("quickstart").storage(S3Storage("your-state-bucket"))
src = cfg.data_source(
    "sample",
    DataSourceS3(s3_bucket="skippr-public-sample-data", s3_prefix="bike-hire"),
)
warehouse = cfg.data_sink(
    "warehouse",
    DataSinkSnowflake(
        account="myorg-myaccount",
        user="skippr_loader",
        private_key_path="${SNOWFLAKE_PRIVATE_KEY_PATH}",
        warehouse="COMPUTE_WH",
        database="RAW_DATA",
        schema="PUBLIC",
        role="LOADER_ROLE",
        stage="@SKIPPR_STAGE",
    ),
)
cfg.pipeline("bikehire", Pipeline(data_source=src, data_sink=warehouse))
cfg.save("skippr.yml")

CLI

skipprd --workspace quickstart --storage-mode s3 --skippr-s3-bucket your-state-bucket \
  connect data-source s3 \
  --pipeline bikehire \
  --name sample \
  --s3-bucket skippr-public-sample-data \
  --s3-prefix bike-hire

skipprd connect data-sink snowflake \
  --pipeline bikehire \
  --name warehouse \
  --account myorg-myaccount \
  --user skippr_loader \
  --private-key-path '${SNOWFLAKE_PRIVATE_KEY_PATH}' \
  --warehouse COMPUTE_WH \
  --database RAW_DATA \
  --schema PUBLIC \
  --role LOADER_ROLE \
  --stage '@SKIPPR_STAGE'

skippr.yml

skippr:
  workspace: quickstart
  skippr_s3_bucket: your-state-bucket
  skipprd_el_storage_mode: s3

pipelines:
  bikehire:
    data_source: data_sources.sample
    data_sink: data_sinks.warehouse

data_sources:
  sample:
    S3:
      s3_bucket: skippr-public-sample-data
      s3_prefix: bike-hire

data_sinks:
  warehouse:
    Snowflake:
      account: "myorg-myaccount"
      user: "skippr_loader"
      private_key_path: "${SNOWFLAKE_PRIVATE_KEY_PATH}"
      warehouse: "COMPUTE_WH"
      database: "RAW_DATA"
      schema: "PUBLIC"
      role: "LOADER_ROLE"
      stage: "@SKIPPR_STAGE"

Pipeline state lives in your-state-bucket: S3Storage("your-state-bucket") in Python, --storage-mode s3 --skippr-s3-bucket on the CLI. LocalStorage() / --storage-mode local keeps it on disk.

Sync

Python

import skippr

s = skippr.Session(skippr.Config.discover().get_pipeline("bikehire"))
s.doctor()
s.discover()
s.sync(once=True)
s.df().to_pandas()

CLI

skipprd doctor
skipprd discover --pipeline bikehire --log
skipprd sync --pipeline bikehire --once --log
skipprd df --pipeline bikehire

df() is the engine view: the WAL, since Snowflake is not an Iceberg sink. To prove the destination table:

select count(*) from RAW_DATA.PUBLIC.bikehire;

Table names are the pipeline namespace, lowercased, dots turned into underscores.

dbt source

dbt reads the warehouse table, not the WAL.

version: 2

sources:
  - name: skipprd
    database: RAW_DATA
    schema: PUBLIC
    tables:
      - name: bikehire

Then model it and test it.