Stage, formati file e COPY INTO

Automazione delle pipeline di dati in Snowflake

Emily Melhuish

Technical Curriculum Developer, Snowflake

Snowflake alimenta molti workload

Screenshot 2026-05-11 at 11.52.34 am.png

¹ * Snowflake Learning Material

Screenshot 2026-05-11 at 10.48.51 am.png

¹ * Snowflake Learning Material

Logo fittizio di Harbr

Su Harbr

Piattaforma globale per logistica e supply chain
Usa Snowflake per tracciare spedizioni, inventario di magazzino e performance di consegna
Acquisisce dati dai fornitori in più regioni

Diagramma di caricamento dati

Stage = area di storage temporanea

Due tipi di stage

Stage interni

Stage esterno

Directory Table

CREATE STAGE harbr_stage
  DIRECTORY = (ENABLE = TRUE)

Crittografia stage interni

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

Crittografia stage esterni

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

ENCRYPTION = 
([ TYPE = 'AWS_SSE_S3' ]) 

...

Formati strutturati: CSV con delimitatori, righe header, regole di quoting
Formati semi-strutturati: JSON, Parquet, Avro, ORC e XML.
- Carica direttamente in una colonna VARIANT
- Definisci una volta le regole di parsing, riusale in tutti i load
Aggiorna una sola volta l’oggetto formato

SQL

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

Carica da uno stage denominato usando un file format denominato

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

Opzioni ON_ERROR	Comportamento
`ABORT_STATEMENT`	Interrompe tutto al primo errore (default)
`CONTINUE`	Salta le righe errate, carica il resto
`SKIP_FILE`	Salta l’intero file se contiene errori
`SKIP_FILE_<num>`	Salta il file solo se gli errori superano una soglia
`SKIP_FILE_<num>%`	Salta il file solo se gli errori superano una soglia in %

Valida senza caricare = dry run

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

Ispeziona la cronologia dei load

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

Automazione delle pipeline di dati in Snowflake