Stages, Dateiformate und COPY INTO

Automatisierung von Datenpipelines in Snowflake

Emily Melhuish

Technical Curriculum Developer, Snowflake

Snowflake treibt viele Workloads an

Screenshot 2026-05-11 at 11.52.34 am.png

¹ * Snowflake Lernmaterial

Screenshot 2026-05-11 at 10.48.51 am.png

¹ * Snowflake Lernmaterial

Fake-Logo für Harbr

Über Harbr

Datenlade-Diagramm

Stage = temporärer Speicherort

Zwei Stagetypen

Interne Stages

Externe Stage

Directory Table

CREATE STAGE harbr_stage
  DIRECTORY = (ENABLE = TRUE)

Verschlüsselung für interne Stages

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

Verschlüsselung für externe Stages

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

ENCRYPTION = 
([ TYPE = 'AWS_SSE_S3' ]) 

...

Strukturierte Formate: CSV mit bestimmten Trennzeichen, Kopfzeilen, Anführungsregeln
Semistrukturierte Formate: JSON, Parquet, Avro, ORC und XML
- Direkt in eine VARIANT-Spalte laden
- Parsing-Regeln einmal definieren, bei allen Ladevorgängen wiederverwenden
Formatobjekt einmal aktualisieren

SQL

CREATE FILE FORMAT harbr_csv_format
  TYPE = 'CSV'
  FIELD_DELIMITER = ','
  SKIP_HEADER = 1;

Aus einer benannten Stage mit benanntem Dateiformat laden

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

ON_ERROR-Optionen	Verhalten
`ABORT_STATEMENT`	Gesamten Ladevorgang beim ersten Fehler abbrechen (Standard)
`CONTINUE`	Fehlerhafte Zeilen überspringen, Rest laden
`SKIP_FILE`	Ganze Datei überspringen, wenn sie Fehler enthält
`SKIP_FILE_<num>`	Datei nur überspringen, wenn Fehler eine Anzahl überschreiten
`SKIP_FILE_<num>%`	Datei nur überspringen, wenn Fehler einen %-Schwellenwert überschreiten

Validieren ohne zu laden = Dry Run

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

Ladehistorie prüfen

SELECT * FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
  TABLE_NAME => 'shipments',
  START_TIME => DATEADD('hour', -24,
    CURRENT_TIMESTAMP())));

Automatisierung von Datenpipelines in Snowflake