Görevler ve DAG Düzenleme

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

Emily Melhuish

Technical Curriculum Developer, Snowflake

Manuel İş Yüklerine Çözüm: Görevler

Ekip her sabah SQL betiğini elle çalıştırıyor
Bir çalıştırma kaçarsa görünürlük kaybolur
Snowflake görevleri bunu tamamen otomatikleştirir

Örnek:

-- Bugünün manuel rutini (kırılgan)
CALL logistics.refresh_ops_dashboard();

Görevler Hakkında Bilmemiz Gerekenler

Görevler zamanlanmış SQL çalıştırır — harici CRON gerekmez
SQL deyimleri çalıştırır veya saklı yordam çağırır
Hesaplama ve zamanlama Snowflake’te yereldir

CREATE TASK logistics.refresh_dashboard
  WAREHOUSE = harbr_wh
  SCHEDULE = 'USING CRON 0 6 * * * UTC'
AS CALL logistics.refresh_ops_dashboard();

Bağımsız Görev Oluşturma

CREATE OR REPLACE TASK transform_delivery_summary 
  WAREHOUSE = harbr_wh 
  SCHEDULE = 'USING CRON 5 * * * * UTC' 
  AS INSERT INTO delivery_summary 

SELECT shipment_id, status, updated_at 
FROM delivery_events 
WHERE processed = FALSE;

Depo Tabanlı vs Sunucusuz

Depo tabanlı

Adlandırılmış sanal depo kullanır
Çalıştırma başına minimum 60 sn faturalama

CREATE TASK my_task
  WAREHOUSE = harbr_wh  -- named warehouse
  SCHEDULE = '5 MINUTE'
AS INSERT INTO ...;

Sunucusuz

WAREHOUSE ifadesini çıkarın
Tüketilen saniye başına faturalama; boşta maliyet yok

CREATE TASK my_serverless_task
  -- no WAREHOUSE clause
  SCHEDULE = '5 MINUTE'
AS INSERT INTO ...;

DAG Tabanlı Görev Düzenleme

DAG diyagramı — ingest_raw (kök görev, saat simgesi) → clean_events (ingest_raw sonrası çalışır) → build_summary (clean_events sonrası çalışır)

DAG: Yönlendirilmiş Çevrimsiz Grafik
Kök görev CRON zamanlamasını tutar
Alt görevler öncülünü AFTER ile belirtir
Tüm zincir merkezîdir
DAG’ler ne zaman ve hangi sırayla çalışılacağını kontrol eder

Görev Durumlarını Yönetme

Durum diyagramı — SUSPENDED → STARTED → SUCCEEDED veya FAILED

-- Bir görevi etkinleştirme
ALTER TASK mytask RESUME; 
-- Bir görevi duraklatma
ALTER TASK mytask SUSPEND; 
-- Bir görevi yürütme
EXECUTE TASK mytask SUSPEND;

Görevler ve Akışların Birleşimi

CDC boru hattı diyagramı — delivery_events → akış değişiklikleri yakalar → akışta veri var mı? EVET → görev tetiklenir → delivery_summary güncellenir / HAYIR → görev atlar

CREATE OR REPLACE TASK logistics.process_delivery_events
    WAREHOUSE = compute_wh
    SCHEDULE = '5 minute'
    WHEN SYSTEM$STREAM_HAS_DATA('logistics.delivery_events_stream')
AS
INSERT INTO logistics.processed_events
SELECT * FROM logistics.delivery_events_stream;

Hadi pratik yapalım!

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