Etapy, formaty plików i COPY INTO

Automatyzacja potoków danych w Snowflake

Emily Melhuish

Technical Curriculum Developer, Snowflake

Snowflake obsługuje wiele typów obciążeń

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

Fikcyjne logo Harbr

O Harbr

Diagram ładowania danych

Etap = tymczasowa lokalizacja przechowywania

Dwa typy etapów

Etapy wewnętrzne

Etap zewnętrzny

Tabela katalogowa

CREATE STAGE harbr_stage
  DIRECTORY = (ENABLE = TRUE)

Szyfrowanie etapu wewnętrznego

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

Szyfrowanie etapu zewnętrznego

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

ENCRYPTION = 
([ TYPE = 'AWS_SSE_S3' ]) 

...

Formaty strukturalne: CSV z określonymi separatorami, nagłówkami i regułami cytowania
Formaty częściowo strukturalne: JSON, Parquet, Avro, ORC i XML.
- Ładowanie bezpośrednio do kolumny VARIANT
- Definiowanie reguł parsowania raz, wielokrotne używanie
Aktualizacja obiektu formatu jednorazowo

SQL

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

Ładowanie z nazwanego etapu przy użyciu nazwanego formatu pliku

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

Opcje ON_ERROR	Zachowanie
`ABORT_STATEMENT`	Przerwanie całego ładowania przy pierwszym błędzie (domyślnie)
`CONTINUE`	Pomijanie błędnych wierszy, ładowanie pozostałych
`SKIP_FILE`	Pomijanie całego pliku w razie błędów
`SKIP_FILE_<num>`	Pomijanie pliku, gdy liczba błędów przekroczy limit
`SKIP_FILE_<num>%`	Pomijanie pliku, gdy odsetek błędów przekroczy próg

Walidacja bez ładowania = próbne uruchomienie

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

Sprawdzanie historii ładowania

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

Automatyzacja potoków danych w Snowflake