İş akışı orkestrasyonu: cron, DAG ve Airflow

Neden önemli?

Veri boru hattı (pipeline) dersinde tek bir ETL betiği yazdın. Gerçek kurumda ise onlarca iş birbirine bağlıdır: Anadolu Market'te her gece 02:00'de siparişler çekilir, temizlenir, stok tablosu güncellenir, şehir raporu üretilir, yöneticilere e-posta gider. Ilgaz Bank'ta gün sonu mutabakatı, tüm şubelerin işlem dosyaları gelmeden başlayamaz. Bu işleri "doğru sırada, doğru zamanda, hata olursa yeniden deneyerek ve kimse bakmasa da haber vererek" çalıştırma işine orkestrasyon (orchestration) denir. Bu derste önce temelleri (cron, DAG, idempotentlik) saf Python ile kuracak, sonra Airflow ve benzeri araçların bunları nasıl paketlediğini göreceksin.

Cron: ne zaman çalışsın?

Cron, Unix sistemlerinin zamanlayıcısıdır; sözdizimi Airflow, Kubernetes CronJob, GitHub Actions ve bulut zamanlayıcılarında da kullanılır. Bir ifade beş alandan oluşur:

# Linux / macOS: crontab -e
0 2 * * *  cd /opt/anadolu && .venv/bin/python -m etl.gunluk >> /var/log/etl.log 2>&1
# Windows'ta Görev Zamanlayıcı (Task Scheduler) ya da schtasks:
schtasks /create /tn "AnadoluETL" /tr "C:\anadolu\.venv\Scripts\python.exe -m etl.gunluk" /sc daily /st 02:00

Bir alan nasıl açılır? */15 "0'dan başlayıp 15'er adım", 1-5 "aralık", 1,3,5 "liste" demektir; 10-30/10 ise aralık içinde adım. Aşağıdaki kutu bir alanı değer kümesine çeviriyor; alıştırmalarda bunun üzerine "bir sonraki çalışma zamanı" hesaplayıcısı kuracaksın.

def alan_ac(alan, alt, ust):
    degerler = set()
    for parca in alan.split(","):
        aralik, _, adim = parca.partition("/")
        if aralik == "*":
            bas, son = alt, ust
        elif "-" in aralik:
            bas, son = map(int, aralik.split("-"))
        else:
            bas = son = int(aralik)
        degerler.update(range(bas, son + 1, int(adim) if adim else 1))
    return sorted(degerler)

print(alan_ac("*/15", 0, 59))
print(alan_ac("1-5", 0, 6))
print(alan_ac("8,12,18", 0, 23))
print(alan_ac("10-30/10", 0, 59))

Python içinden zamanlama gerekiyorsa iki popüler paket vardır (tarayıcıda çalışmaz, kendi bilgisayarında dene):

# pip install schedule apscheduler
import schedule, time
schedule.every().day.at("02:00").do(gunluk_etl)
schedule.every(15).minutes.do(stok_kontrol)
while True:
    schedule.run_pending()
    time.sleep(30)

from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
zamanlayici = BlockingScheduler(timezone="Europe/Istanbul")
zamanlayici.add_job(gunluk_etl, CronTrigger.from_crontab("0 2 * * *"), max_instances=1)
zamanlayici.start()

DAG: hangi sırayla çalışsın?

Cron "ne zaman" sorusunu çözer, ama "temizlik bitmeden rapor başlamasın" diyemez. Görevler ve aralarındaki bağımlılıklar bir yönlü döngüsüz çizge (Directed Acyclic Graph, DAG) oluşturur: oklar "önce bu, sonra şu" der, döngü olamaz (A, B'yi; B de A'yı bekliyorsa hiçbiri başlayamaz). Bir DAG'ı geçerli bir sıraya dizmeye topolojik sıralama denir ve Python'da standart kütüphanede hazırdır: graphlib.TopologicalSorter.

from graphlib import TopologicalSorter, CycleError

# görev: {önce bitmesi gerekenler}
akis = {
    "siparis_cek": set(),
    "musteri_cek": set(),
    "temizle": {"siparis_cek", "musteri_cek"},
    "stok_guncelle": {"temizle"},
    "sehir_raporu": {"temizle"},
    "eposta_gonder": {"sehir_raporu", "stok_guncelle"},
}
print(list(TopologicalSorter(akis).static_order()))

# paralel katmanlar: aynı anda çalışabilecek görevler
ts = TopologicalSorter(akis)
ts.prepare()
katman = 1
while ts.is_active():
    hazir = sorted(ts.get_ready())
    print(f"Katman {katman}:", hazir)
    ts.done(*hazir)
    katman += 1

try:
    list(TopologicalSorter({"a": {"b"}, "b": {"c"}, "c": {"a"}}).static_order())
except CycleError as e:
    print("Döngü:", e.args[1])

get_ready() bağımlılıkları bitmiş görevleri verir; done() ile bitenleri bildirdikçe yenileri açılır. Orkestratörlerin çalıştırıcıları (executor) özünde bu döngüyü işçi havuzuyla çalıştırır.

Airflow: endüstri standardı

Apache Airflow, DAG'ları Python koduyla tanımladığın, zamanlayan, çalıştıran, yeniden deneyen ve web arayüzünde izleyen açık kaynak bir platformdur. Temel kavramlar:

# pip install apache-airflow  (yerelde: airflow standalone)
from datetime import datetime, timedelta
from airflow.decorators import dag, task
from airflow.operators.bash import BashOperator

varsayilan = {"retries": 3, "retry_delay": timedelta(minutes=5),
              "retry_exponential_backoff": True, "email_on_failure": True}

@dag(schedule="0 2 * * *", start_date=datetime(2025, 1, 1), catchup=False,
     default_args=varsayilan, tags=["anadolu-market"])
def gunluk_satis():
    @task
    def siparis_cek(ds=None) -> str:          # ds: işlenen gün (YYYY-MM-DD)
        yol = f"/data/ham/siparis_{ds}.csv"
        # ... API'den çek, dosyaya yaz ...
        return yol                            # XCom ile sonraki göreve gider

    @task
    def temizle(yol: str) -> int:
        # ... oku, doğrula, temiz tabloya yaz ...
        return 1250                           # satır sayısı

    @task
    def rapor(satir: int):
        print(f"{satir} satır işlendi")

    yedek = BashOperator(task_id="yedekle", bash_command="pg_dump anadolu > /yedek/{{ ds }}.sql")
    satir = temizle(siparis_cek())
    satir >> [rapor(satir), yedek]

gunluk_satis()

TaskFlow API'de (@task) bir görevin dönüş değerini diğerine argüman olarak vermek hem bağımlılığı hem XCom aktarımını kendiliğinden kurar. {{ ds }} Jinja şablonudur; Airflow onu çalışmanın tarihiyle doldurur. catchup=True olsaydı start_date'ten bugüne kadar kaçırılan tüm günler tek tek çalıştırılırdı (backfill).

İdempotentlik ve yeniden çalıştırma

Orkestrasyonun altın kuralı: bir görevi aynı girdiyle iki kez çalıştırmak, bir kez çalıştırmakla aynı sonucu vermelidir (idempotent). Çünkü görevler yeniden denenir, elle yeniden başlatılır, backfill edilir. "Satırları tabloya ekle" idempotent değildir: ikinci çalışmada her satır iki kez olur. "Bu günün bölümünü (partition) sil, sonra yaz" ya da "anahtar varsa güncelle" (upsert) idempotenttir.

import sqlite3
db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE gunluk_satis (tarih TEXT, sehir TEXT, ciro REAL)")

def yukle(tarih, satirlar):
    with db:   # işlem (transaction): ya hepsi ya hiçbiri
        db.execute("DELETE FROM gunluk_satis WHERE tarih = ?", (tarih,))
        db.executemany("INSERT INTO gunluk_satis VALUES (?, ?, ?)",
                       [(tarih, s, c) for s, c in satirlar])

veri = [("İzmir", 1500.0), ("Ankara", 900.0)]
yukle("2025-03-01", veri)
yukle("2025-03-01", veri)          # yeniden deneme: kopya oluşmaz
yukle("2025-03-02", [("İzmir", 700.0)])
print(db.execute("SELECT tarih, COUNT(*), SUM(ciro) FROM gunluk_satis GROUP BY tarih").fetchall())

Diğer ilkeler: görev işlediği zaman dilimini parametre olarak alsın (datetime.now() değil; yoksa dünün işini bugün yeniden çalıştıramazsın), çıktı dosyaları tarihli olsun, yan etkiler (e-posta, ödeme) tekrar gönderimi önleyen bir anahtarla korunsun.

Yeniden deneme, hata bildirimi ve gözlemlenebilirlik

Ağ kopması, kilitli tablo, geçici 503 gibi geçici hatalar yeniden denenir; bekleme süresi her denemede katlanır (üstel geri çekilme, exponential backoff) ki zaten zorlanan sisteme yük bindirilmesin. Kalıcı hatalar (bozuk veri, yanlış parola) yeniden denenmez; hemen başarısız olup haber verilir.

import functools

def yeniden_dene(deneme=3, ilk_bekleme=1.0, gecici=(ConnectionError, TimeoutError), uyu=None):
    def dekorator(f):
        @functools.wraps(f)
        def sarmal(*args, **kwargs):
            bekleme = ilk_bekleme
            for i in range(1, deneme + 1):
                try:
                    return f(*args, **kwargs)
                except gecici as e:
                    if i == deneme:
                        raise
                    print(f"  {f.__name__} deneme {i} başarısız ({e}); {bekleme} sn sonra tekrar")
                    (uyu or (lambda s: None))(bekleme)
                    bekleme *= 2
        return sarmal
    return dekorator

sayac = {"n": 0}

@yeniden_dene(deneme=4)
def siparis_cek():
    sayac["n"] += 1
    if sayac["n"] < 3:
        raise ConnectionError("API yanıt vermedi")
    return 120

print("Çekilen satır:", siparis_cek())

Haber verme katmanları: görev başarısız olduğunda (Airflow'da on_failure_callback) Slack/Teams/e-posta mesajı; iş hiç başlamazsa sessiz kalmamak için "ölü adam anahtarı" (dead man's switch: iş her başarıyla bittiğinde bir izleme servisine sinyal gönderir, sinyal gelmezse alarm çalar); SLA ("rapor 07:00'ye kadar hazır olmalı") ihlalinde uyarı.

Gözlemlenebilirlik (observability) için her görev en azından şunları yapılandırılmış log (JSON satırı) olarak yazmalı: görev adı, çalışma kimliği, işlenen tarih, başlangıç-bitiş, süre, okunan/yazılan satır sayısı, sonuç. Bu sayılar metrik sistemine (Prometheus, Datadog, CloudWatch) gider; "dün 1250 satır geldi, bugün 3" gibi sessiz veri sorunları ancak böyle yakalanır.

import json, time

def calistir_ve_logla(ad, f, tarih):
    bas = time.perf_counter()
    kayit = {"gorev": ad, "tarih": tarih}
    try:
        kayit["satir"] = f(tarih)
        kayit["durum"] = "basarili"
    except Exception as e:
        kayit.update(durum="basarisiz", hata=f"{type(e).__name__}: {e}")
    kayit["sure_ms"] = round((time.perf_counter() - bas) * 1000)
    print(json.dumps({k: v for k, v in kayit.items() if k != "sure_ms"}, ensure_ascii=False))
    return kayit

calistir_ve_logla("temizle", lambda t: 1250, "2025-03-01")
calistir_ve_logla("rapor", lambda t: 1 / 0, "2025-03-01")

import veri
from datetime import date, timedelta
conn = veri.baglan("lezzet_kurye", tablolar=["lk_orders"])

def gunluk_siparis(ds: date) -> int:           # görev: yalnız kendi dilimine bakar
    bas, bit = ds.isoformat(), (ds + timedelta(days=1)).isoformat()
    return conn.execute("SELECT COUNT(*) FROM lk_orders WHERE placed_at >= ? AND placed_at < ?",
                        (bas, bit)).fetchone()[0]

ds = date(2024, 10, 1)                         # backfill: 1-7 Ekim
while ds <= date(2024, 10, 7):
    print(ds, gunluk_siparis(ds))
    ds += timedelta(days=1)

from airflow.decorators import dag, task      # Airflow 3: from airflow.sdk import dag, task
from datetime import datetime

@dag(schedule="@daily", start_date=datetime(2024, 10, 1), catchup=False,
     max_active_runs=1, default_args={"retries": 2})
def lezzet_gunluk():
    @task
    def subeler() -> list[str]:
        return ["izmir", "ankara", "istanbul"]

    @task
    def cek(sube: str, data_interval_start=None) -> str:   # bağlam argümanı adla enjekte edilir
        ds = data_interval_start.strftime("%Y-%m-%d")
        return f"s3://ham/{sube}/{ds}.parquet"              # XCom'a veri değil YOL koy

    @task(multiple_outputs=True)
    def ozet(yollar: list[str]) -> dict:
        return {"dosya": len(yollar), "durum": "tamam"}

    yollar = cek.expand(sube=subeler())       # dinamik eşleme: her şube için bir görev örneği
    ozet(yollar)

lezzet_gunluk()

airflow dags backfill lezzet_gunluk -s 2024-10-01 -e 2024-10-07       # Airflow 2
airflow backfill create --dag-id lezzet_gunluk \
    --from-date 2024-10-01 --to-date 2024-10-07 --max-active-runs 2   # Airflow 3

from airflow.sdk import Asset, dag, task
siparisler = Asset("s3://ambar/siparisler")
kuryeler = Asset("s3://ambar/kuryeler")

@dag(schedule="@hourly")
def uretici():
    @task(outlets=[siparisler])
    def yaz(): ...
    yaz()

@dag(schedule=[siparisler, kuryeler])        # ikisi de güncellenince
def kurye_performans():
    ...

# Prefect: sıradan Python, @flow ve @task
from prefect import flow, task
@task(retries=2)
def cek(sube: str, gun: str) -> str: ...
@flow
def lezzet_gunluk(gun: str):
    yollar = cek.map(["izmir", "ankara"], gun)

# Dagster: varlık (asset) odaklı; "hangi tablo neyden üretilir"
from dagster import asset, DailyPartitionsDefinition
gunluk = DailyPartitionsDefinition(start_date="2024-10-01")
@asset(partitions_def=gunluk)
def ham_siparis(context): ...
@asset(partitions_def=gunluk)
def gunluk_ozet(ham_siparis): ...           # bağımlılık parametre adından çıkar