Etape, formate de fișiere și COPY INTO

Automatizarea pipeline-urilor de date în Snowflake

Emily Melhuish

Technical Curriculum Developer, Snowflake

Snowflake susține multe tipuri de sarcini

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 fictiv pentru Harbr

Despre Harbr

Platformă globală de logistică și lanț de aprovizionare
Utilizează Snowflake pentru urmărirea expediențelor, inventarului și performanței livrărilor
Ingresează date din sistemele furnizorilor din mai multe regiuni

Diagramă de încărcare a datelor

Etapă = locație temporară de stocare

Două tipuri de etape

Etape interne

Etapă externă

Tabel director

CREATE STAGE harbr_stage
  DIRECTORY = (ENABLE = TRUE)

Criptare etapă internă

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

Criptare etapă externă

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

ENCRYPTION = 
([ TYPE = 'AWS_SSE_S3' ]) 

...

Formate structurate: CSV cu delimitatori, rânduri de antet, reguli de cotare
Formate semi-structurate: JSON, Parquet, Avro, ORC și XML.
- Încărcare directă într-o coloană VARIANT
- Definiți regulile de parsare o singură dată, reutilizați la orice încărcare
Actualizați obiectul de format o singură dată

SQL

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

Încărcare dintr-o etapă numită cu un format de fișier numit

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

Opțiuni ON_ERROR	Comportament
`ABORT_STATEMENT`	Oprește încărcarea la prima eroare (implicit)
`CONTINUE`	Ignoră rândurile invalide, încarcă restul
`SKIP_FILE`	Ignoră fișierul dacă conține erori
`SKIP_FILE_<num>`	Ignoră fișierul dacă erorile depășesc un număr specificat
`SKIP_FILE_<num>%`	Ignoră fișierul dacă erorile depășesc un procent specificat

Validare fără încărcare = rulare de test

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

Inspecție istoric încărcare

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

Automatizarea pipeline-urilor de date în Snowflake