Airflow'da ilk DAG'dan düzenli iş akışına: yazım rehberi

DAG nedir ve bir DAG dosyası neyi içerir?

Apache Airflow'da iş akışı, DAG (yönlü döngüsüz çizge) olarak tanımlanır. DAG, görevleri (task) ve aralarındaki bağımlılıkları tanımlayan Python kodudur. "Yönlü" bağımlılığın bir yönü olduğunu, "döngüsüz" ise bir görevin kendi çıktısına geri dönemeyeceğini anlatır. Airflow bu dosyayı okuyup görevleri doğru sırayla, doğru zamanda çalıştırır.

Bir DAG dosyasının üç ana parçası vardır: DAG'ın kimliği ve zamanlaması, görevler ve görevlerin sırası. Dosya sadece iş akışını tanımlar; ağır işi görevler yapar. Bu ayrım önemlidir, çünkü Airflow DAG dosyasını sürekli tekrar okur (parse eder). Dosyanın üst düzeyinde veritabanı sorgusu, API çağrısı ya da büyük hesaplama bulunması, zamanlayıcıyı yavaşlatan klasik hatadır. Ağır iş yalnız görevin içinde çalışmalıdır.

Airflow'un görev tanımlama yolları iki ailedir: operatörler (hazır görev türleri) ve TaskFlow API'nin @task dekoratörü. Operatör, belirli bir işi yapan hazır sınıftır; örneğin komut çalıştıran ya da boş görev oluşturan operatörler vardır. @task ise sıradan bir Python fonksiyonunu görev haline getirir. Yeni yazılan Python ağırlıklı akışlarda ikincisi daha okunaklıdır.

İlk DAG: tanımlamanın üç yolu

Resmî belgeler bir DAG'ı üç şekilde tanımlamayı gösterir: bağlam yöneticisi (with), kurucu çağrısı ve @dag dekoratörü. En yaygını bağlam yöneticisidir; with bloğunun içinde oluşturulan her görev DAG'a eklenir.

from airflow.sdk import DAG from airflow.providers.standard.operators.empty import EmptyOperator

with DAG( dag_id="example_dag", start_date=START_DATE, schedule="@daily", ): EmptyOperator(task_id="task")

Parametreler şunları söyler: dag_id akışın benzersiz adıdır ve arayüzde bu ad görünür. start_date, zamanlamanın hangi tarihten itibaren hesaplanacağını belirler; sabit ve geçmiş bir tarih olmalıdır. schedule, ne sıklıkla çalışacağını söyler.

@dag dekoratörü ile yazarsanız dikkat edilecek nokta vardır: dekoratörlü fonksiyonu sonunda çağırmanız ve sonucu üst düzeyde bir nesneye atamanız gerekir; aksi halde Airflow DAG'ı yüklemez. Belgeler bunu özellikle vurgular.

from airflow.sdk import dag

@dag(start_date=START_DATE, schedule="@daily") def pipeline(): EmptyOperator(task_id="task")

pipeline()

Belgedeki sürüme göre içe aktarma yolları değişebilir. Kurulu Airflow sürümünüzün belgesindeki örneği esas alın; eski sürümlerde DAG sınıfı farklı bir modülden içe aktarılır.

Görevler arası bağımlılık nasıl tanımlanır?

Bağımlılık, görevlerin hangi sırayla çalışacağını belirler. Belgelerin önerdiği yol >> ve << operatörleridir. Soldaki görev bitince sağdaki başlar:

extract >> transform >> load extract >> [check_a, check_b]

İkinci satır, çekme görevinden sonra kontrol görevlerinin paralel çalışmasını sağlar (fan-out). Tersine, birden çok görev bittikten sonra tek bir göreve geçmek (fan-in) için listeyi sola koyarsınız. Uzun ve doğrusal bir zincir için chain yardımcısı aynı işi sade biçimde yapar:

from airflow.sdk import chain chain(extract, transform, check, load)

Bağımlılığı doğru kurmak, hata yönetimi açısından da önemlidir. Varsayılan davranışta bir görev başarısız olursa ondan sonra gelenler çalışmaz. Gerçek bir iş akışında bu bazen istenmez; örneğin bir bildirim görevi, önceki görevler başarısız olsa da çalışmalıdır. Bu durum görevin tetikleme kuralıyla (trigger rule) ayarlanır; konu işleri başlatmayı değil, ne zaman başlatılacağını belirler.

Bir iş akışı çizerken kağıt üzerinde önce kutuları ve okları çizmek zaman kazandırır: hangi görevler bağımsız, hangileri aynı çıktıya bağlı, hangi noktada ayrışma ve birleşme olacak. Kod, çizimin doğrudan çevirisi olmalıdır.

Zamanlama: schedule ve start_date nasıl çalışır?

schedule parametresi birkaç biçimde verilebilir. Belgelerde şu örnekler yer alır: "@daily" gibi hazır ön ayarlar, cron ifadesi, bir kez çalışması için "@once" ve sürekli çalışma için "@continuous". Zamanlama vermek de zorunlu değildir; DAG elle ya da API ile tetiklenebilir.

Yeni başlayanları en çok şaşırtan nokta, Airflow'un bir çalışmayı, ait olduğu aralık bittikten sonra başlatmasıdır. Günlük bir DAG'ın belirli bir günü işleyen çalışması, o günün bitiminde tetiklenir. Bu yüzden "bugünün verisini bugün işlesin" beklentisi ilk başta yanlış sonuç doğurur. Çalışmanın işlediği veri aralığını, görev içinde Airflow'un sağladığı tarih bağlamından (context) okumak gerekir; kod içinde güncel saati elle almak (datetime.now gibi) bu yüzden tehlikelidir, çünkü aynı çalışmayı sonradan yeniden koştuğunuzda farklı veri işler.

Bu davranış doğrudan tekrar çalıştırılabilirlik (idempotency) ile ilgilidir. Geçmiş bir tarihi yeniden çalıştırmak, aynı sonucu üretmelidir. Görevi "belirli aralığın verisini yaz ve varsa üzerine yaz" biçiminde kurmak, ikinci çalıştırmada kayıt çoğalmasını önler.

Geçmiş çalışmaların telafisi (catchup), kritik bir ayardır. Kapalı olduğu dönemde kaçırılan aralıkların toplu çalıştırılıp çalıştırılmayacağını belirler. Belgenin bu bölümü ayrıntı vermese de örneklerde catchup=False ifadesi görülür. Yeni bir DAG'ı açarken geçmiş tarihli bir başlangıçla yüzlerce çalışma tetiklemek istemiyorsanız bu ayarı bilinçli seçin.

TaskFlow API: görevler arasında veri aktarmak

@task dekoratörü, bağımsız fonksiyonları görev yapar ve bir görevin döndürdüğü değeri sonrakine parametre olarak geçirmenizi sağlar. Airflow bu değer aktarımını XCom üzerinden arka planda yönetir.

from airflow.sdk import dag, task

@dag(start_date=START_DATE, schedule="@daily") def siparis_akisi(): @task def siparisleri_al() -> list: return kaynaktan_oku()

@task def toplami_hesapla(kayitlar: list) -> int: return sum(k["tutar"] for k in kayitlar)

toplami_hesapla(siparisleri_al())

siparis_akisi()

Burada bağımlılık da otomatik oluşur: toplami_hesapla, siparisleri_al'ın çıktısını kullandığı için ondan sonra çalışır.

Önemli sınır: XCom küçük veriler içindir (kimlikler, dosya yolları, sayılar). Büyük bir tablo ya da dosya içeriğini görevler arasında XCom'la taşımak veritabanını şişirir ve akışı yavaşlatır. Büyük veri ortak bir depoya (nesne depolama, veritabanı tablosu) yazılır, görevler arasında yalnızca o verinin konumu aktarılır.

Hangi yolu seçmeli? Python mantığı ağırsa @task, hazır bir entegrasyon varsa (örneğin bir bulut hizmetini çağıran operatör) operatör kullanın. İkisi aynı DAG içinde bir arada çalışabilir.

Sık yapılan yazım hataları ve denetim listesi

• DAG dosyasının üst düzeyinde ağır iş yapmak. Dosya sık sık okunduğu için bu, zamanlayıcıyı yavaşlatır. Bağlantı açma, sorgu çalıştırma gibi işleri görev içine alın.
• start_date için dinamik bir değer (örneğin şimdiki zaman) kullanmak. Sabit ve geçmiş bir tarih verin.
• Aynı dag_id'yi birden çok dosyada kullanmak.
• Görevleri tekrar çalıştırılabilir yazmamak. Yeniden koşma, çift kayıt üretmemeli.
• Parolaları ve erişim anahtarlarını kodun içine yazmak. Airflow'un bağlantı (connection) ve değişken (variable) yapıları ya da bir gizli bilgi deposu kullanın.
• XCom ile büyük veri taşımak.
• Bağımlılıkları kodda değil, kafada tutmak. Arayüzdeki çizge görünümünü açıp akışın çizdiğiniz şemayla aynı olduğunu doğrulayın.
• Test etmeden yayınlamak. Önce yerelde DAG'ın hatasız yüklendiğini, ardından tek tek görevleri deneyin.

Başlangıç, orta ve ileri düzey için yol haritası şöyle kurulur: önce tek DAG, birkaç görev ve doğrusal bağımlılık; sonra TaskFlow, paralel kollar, zamanlama kuralları ve bağlantı yönetimi; en sonda dinamik görev üretimi, üretim ortamı ölçekleme ve izleme. Hata yönetimi (yeniden deneme ve uyarılar) ayrı bir sayfada ayrıntılı anlatılır.

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

Airflow ile Veri Pipeline'ları programı, DAG tasarımını Python tabanlı iş akışları olarak ele alan bir modül içerir; Zamanlama, Bağımlılık ve Tetikleyiciler, Airflow Veri Akışı ve XCom ile Operatör Kütüphanesi ve Entegrasyonlar modülleri de bu sayfadaki başlıkları tamamlar. Program, veri mühendisleri, analitik mühendisleri ve veri hazırlama işini otomatikleştirmek isteyen backend geliştiricileri için uygundur. Kurumsal ekiplerde gece çalışan manuel betikleri izlenebilir, yeniden denenebilir iş akışlarına dönüştürmek isteyen veri ekipleri için çıkış noktası olabilir.