Airflow iş akışında hata nasıl yönetilir: yeniden deneme, zaman aşımı ve uyarı

Hata yönetimi neden akış tasarımının parçasıdır?

Gerçek veri iş akışlarında hata istisna değil, beklenen bir durumdur. Kaynak sistem geçici olarak yanıt vermez, ağ koparsa, dosya geç gelir, bir sorgu zaman aşımına uğrar. Airflow'da hata yönetimi, bu durumların hangisinin kendiliğinden düzelecek geçici hata, hangisinin insan müdahalesi isteyen kalıcı hata olduğunu ayırarak kurulur.

Şu sorulara cevap verilmelidir. Birincisi, hata geçiciyse görev kaç kez ve hangi aralıkla yeniden denensin? İkincisi, görev takılırsa ne kadar beklensin? Üçüncüsü, tüm denemeler tükendiğinde kim, hangi kanaldan haberdar olsun? Bu katmanlar birlikte çalışmazsa ya gereksiz alarm yorgunluğu ya da fark edilmeden bozulan veri oluşur.

Önemli bir ön koşul, görevlerin tekrar çalıştırılabilir yazılmasıdır. Yeniden deneme, aynı işi ikinci kez çalıştırır. Görev "kayıt ekle" biçiminde yazıldıysa her deneme kayıt çoğaltır; "bu aralığın verisini yaz, varsa üzerine yaz" biçiminde yazıldıysa yeniden deneme güvenlidir. Hata yönetimi bu tasarım üzerine kurulur.

Retry kurulumu: retries, retry_delay ve geri çekilme

Airflow belgesine göre varsayılan davranış, hata türüne bakmadan sabit sayıda ve sabit gecikmeyle yeniden denemektir. Temel parametreler:

• retries: görevin başarısız olduktan sonra en fazla kaç kez yeniden deneneceği.
• retry_delay: denemeler arasındaki bekleme süresi (bir timedelta değeri).
• retry_exponential_backoff ve max_retry_delay: bekleme süresinin her denemede büyümesini ve bir üst sınırla kısıtlanmasını sağlayan ayarlar. Geçici yük altındaki dış servislerde ardışık denemeler arasında nefes alma süresi bırakır.

Her görevde ayrı ayrı yazmak yerine DAG düzeyinde default_args ile ortak bir politika tanımlayabilirsiniz; DAG içindeki tüm operatörler bu değerleri devralır.

from datetime import timedelta from airflow.sdk import DAG

with DAG( dag_id="siparis_aktarimi", start_date=BASLANGIC_TARIHI, schedule="@daily", default_args={ "retries": YENIDEN_DENEME_SAYISI, "retry_delay": timedelta(minutes=BEKLEME_DAKIKASI), }, ): ...

Bir görevin ayarını DAG varsayılanından farklı yapmak mümkündür: kısa ömürlü bir hesaplama görevine az deneme, dış API'ye giden göreve daha uzun gecikmeli çok deneme verilebilir. Doğru sayıları bulmak için geçmiş hata kayıtlarına bakın: geçici hatalar çoğunlukla birkaç denemede düzeliyorsa daha fazla deneme sadece gecikmeyi uzatır.

Retry'ın uygun olmadığı durumlar da vardır. Hatalı kimlik bilgisi, bozuk veri biçimi ya da mantık hatası yeniden denemeyle düzelmez. Görev içinden AirflowFailException fırlatmak, kalan deneme hakları olsa bile görevi doğrudan başarısız yapar; AirflowSkipException ise görevi atlandı olarak işaretler.

Zaman aşımı: takılan görevi yakalamak

Hata vermeyen ama bitmeyen görev, hatadan daha tehlikelidir: kaynak tutar, sonraki görevleri bekletir ve kimse fark etmez. execution_timeout parametresi her çalıştırma için bir üst süre koyar; süre aşılırsa Airflow AirflowTaskTimeout hatası üretir, bu hata da retry mekanizmasına girer.

from datetime import timedelta from airflow.providers.standard.operators.empty import EmptyOperator

EmptyOperator( task_id="aktarim", execution_timeout=timedelta(minutes=ZAMAN_ASIMI_DAKIKASI), )

Sensörlerde ek bir zaman aşımı vardır: timeout parametresi, sensörün başarıya ulaşması için tanınan toplam süreyi sınırlar ve yalnızca reschedule modunda anlamlıdır. Belgeye göre süre aşıldığında AirflowSensorTimeout oluşur ve sensör yeniden denenmeden başarısız olur. Yani "dosya gelene kadar bekle" mantığında, dosyanın ne kadar geç gelebileceğini iş tarafıyla netleştirip sensör süresini ona göre vermek gerekir.

Bir de çalışan sürecin kaybolması durumu vardır. Bellek yetersizliği ya da düğümün kaldırılması gibi nedenlerle, çalışıyor görünen ama aslında canlı olmayan görev örnekleri periyodik olarak bulunur, temizlenir ve kalan hakkı varsa yeniden denenir. Bu mekanizmayı bilmek, açıklanamayan başarısızlıklarda bakacağınız yeri gösterir: işçi (worker) kaynakları ve düğüm olayları.

Bildirim: callback'lerle uyarı akışı kurmak

Retry'lar tükendiğinde ya da kritik bir görev başarısız olduğunda ilgili kişilere haber verilmelidir. Airflow bunun için callback'ler sunar: on_failure_callback, on_success_callback ve görev düzeyinde on_retry_callback. Belgeye göre on_failure_callback birden çok fonksiyonu liste olarak kabul eder. Callback fonksiyonu, görev örneği bilgilerini içeren bir context sözlüğü alır.

def notify_failure(context): task = context["task_instance_key_str"] send_message(f"Görev başarısız: {task}")

EmptyOperator( task_id="load", on_failure_callback=[notify_failure], )

Belgede özellikle vurgulanan bir ayrıntı: callback fonksiyonlarındaki hatalar görev günlüklerinde değil, DAG işleyici (dag processor) günlüklerinde görünür. Callback sessizce çalışmıyorsa orada arayın.

Uyarı tasarımında şu ilkeler işe yarar:

• Her deneme için değil, retry'lar tükenince bildirim gönderin. Aksi halde geçici hatalar ekibi yorar. on_retry_callback ise yalnızca izleme amacıyla anlamlıdır.
• Mesaja DAG adı, görev adı, çalışma tarihi ve günlük bağlantısı koyun; kim görürse görsün ne yapacağını anlasın.
• Kritik akışlar için ayrı kanal, bilgilendirme amaçlı akışlar için sakin bir kanal kullanın.
• Callback içinde ağır iş yapmayın; yalnız bildirim gönderin.
• Başarı bildirimi, iş açısından önemli teslimlerde (örneğin sabah raporu hazırsa) faydalıdır, her görevde değil.

Örneğin finans ekibine giden günlük bir rapor DAG'ında son görev başarısızsa hem veri ekibine hem rapor sahibine mesaj gitmeli, raporun gecikeceği bilgisi verilmelidir. Gerçek müdahale için nöbet çizelgesi ve eskalasyon kuralları da tanımlı olmalıdır.

Başarısız görevi nasıl yeniden çalıştırırsınız?

Airflow arayüzünde başarısız görevi seçip Clear dediğinizde yürütücü görevi yeniden çalıştırır. Belgelerde Clear seçenekleri arasında geçmiş, gelecek, yukarı akış, aşağı akış, özyinelemeli ve yalnız başarısız olanlar yer alır. Komut satırında karşılığı airflow tasks clear komutudur ve görev adı deseni ile tarih aralığı verilerek kullanılır.

Geçmiş bir dönemin tümünü yeniden işlemek gerekiyorsa backfill kullanılır; belgeye göre seçilen tarih aralığındaki tüm veri aralıkları için DAG yeniden çalıştırılır. Bunu yapmadan önce görevlerin tekrar çalıştırılabilir olduğundan emin olun.

Düzeltme sırası genelde şudur:

• Önce nedeni bulun: görev günlüğüne bakın. Hata geçici mi, kalıcı mı?
• Kaynak sorunu giderin: bağlantı bilgisi, yetki, kaynak sistem durumu, gelen verinin biçimi.
• Yalnız başarısız görevi ve ona bağlı aşağı akış görevlerini temizleyin; başarılı olanı gereksiz yere yeniden çalıştırmayın.
• Çalışmanın sonuçlarını kontrol edin ve çift kayıt olup olmadığına bakın.
• Aynı hata tekrar edecekse kalıcı çözüm olarak kodu, yeniden deneme ayarını veya zaman aşımını güncelleyin.

Tetikleme kuralı (trigger rule) da hata yönetiminin parçasıdır. Varsayılan olarak bir görev, tüm üst görevleri başarılı olunca çalışır. Bildirim ya da temizlik gibi, önceki görevler başarısız olsa bile çalışması gereken görevler için kuralı değiştirmek gerekir; hangi kuralların bulunduğu Airflow'un DAG belgesindeki ilgili bölümde yer alır.

İzleme alışkanlıkları ve sık hatalar

• Retry'ı sınırsız ya da çok yüksek tutmak. Kalıcı hata saatlerce gizlenir.
• Hata mesajında ayrıntı bırakmamak. Günlüğe giriş yaparken hangi kaynağa, hangi aralık için bağlanıldığını yazın.
• Zaman aşımı koymamak. Takılan görev tüm gece hattını bekletir.
• Tüm hatalara aynı davranmak. Yetki hatasını tekrar tekrar denemek anlamsızdır.
• Callback'in kendisini test etmemek. Alarm kanalı çalışmıyorsa hiç alarm yoktur gibidir.
• Başarısız görevi temizleyip çift kayıt üretmek.
• Önemli akışlarda sessizce atlanan (skipped) görevleri takip etmemek; atlanmak başarı değildir.
• Mesaj yorgunluğu: her küçük hatada uyarı gönderen sistem, gerçek sorunda dikkate alınmaz.

Düzenli olarak şu soruları sorun: Son dönemde en çok hangi görev hata verdi? Hangi hata kendiliğinden düzeldi? Hangisi için insan müdahalesi gerekti? Bu veriler retry sayısını, zaman aşımını ve alarm eşiklerini ayarlamak için değerli bir kaynaktır.

Mindset programlarında bu beceri nerede çalışılır?

Airflow ile Veri Pipeline'ları programındaki İzleme, Hata Yönetimi ve Uyarılar modülü doğrudan bu konuya ayrılmıştır; Zamanlama, Bağımlılık ve Tetikleyiciler ile Üretim Ortamı ve Ölçeklendirme modülleri ise tetikleme kuralları ve çalışma ortamı tarafını tamamlar. Gece çalışan veri hatlarının sahibi olan veri mühendisleri ve nöbet düzenine katılan platform ekipleri için uygundur. Kurumsal ekiplerde, hata durumunda kimin ne yapacağını netleştirmek ve alarm kalitesini artırmak isteyen veri ekipleri için çalışma temeli sağlar.