Aşamalar, Dosya Biçimleri ve COPY INTO

Snowflake'te Veri Pipeline'larını Otomatikleştirme

Emily Melhuish

Technical Curriculum Developer, Snowflake

Snowflake birçok iş yükünü destekler

11.52.34'te ekran görüntüsü 2026-05-11 am.png

¹ * Snowflake Öğrenme Materyali

10.48.51'te ekran görüntüsü 2026-05-11 am.png

¹ * Snowflake Öğrenme Materyali

Harbr için sahte logo

Harbr Hakkında

Küresel lojistik ve tedarik zinciri platformu
Gönderileri, depo stoklarını ve teslimat performansını izlemek için Snowflake kullanır
Birden çok bölgedeki tedarikçi sistemlerinden veri alır

Veri yükleme diyagramı

Aşama = geçici depolama alanı

İki aşama türü

Dahili Aşamalar

Harici Aşama

Dizin Tablosu

CREATE STAGE harbr_stage
  DIRECTORY = (ENABLE = TRUE)

Dahili Aşama Şifreleme

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

Harici Aşama Şifreleme

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

ENCRYPTION = 
([ TYPE = 'AWS_SSE_S3' ]) 

...

Yapılı biçimler: Belirli ayırıcılar, başlık satırları, alıntılama kurallarıyla CSV
Yarı yapılı biçimler: JSON, Parquet, Avro, ORC ve XML.
- Doğrudan VARIANT sütununa yükleyin
- Ayrıştırma kurallarını bir kez tanımlayın, tüm yüklemelerde yeniden kullanın
Biçim nesnesini bir kez güncelleyin

SQL

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

Adlandırılmış aşamadan, adlandırılmış dosya biçimiyle yükleme

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

ON_ERROR Seçenekleri	Davranış
`ABORT_STATEMENT`	İlk hatada tüm yüklemeyi durdurur (öntanımlı)
`CONTINUE`	Hatalı satırları atlar, diğerlerini yükler
`SKIP_FILE`	Herhangi bir hata varsa dosyanın tamamını atlar
`SKIP_FILE_<num>`	Hata sayısı eşik değeri aşarsa dosyayı atlar
`SKIP_FILE_<num>%`	Hata yüzdesi eşik değeri aşarsa dosyayı atlar

Yüklemeden doğrula = deneme çalıştırması

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

Yükleme geçmişini inceleme

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

Snowflake'te Veri Pipeline'larını Otomatikleştirme