Dinamik Görev Eşleme

Airflow ile Veri İş Hatları Oluşturma

Volker Janz

Senior Developer Advocate at Astronomer

Sert kodlanmış görevlerin sorunu

@task
def process_file_1():
    transform("/data/file_1.csv")
@task
def process_file_2():
    transform("/data/file_2.csv")
@task
def process_file_3():
    transform("/data/file_3.csv")

Dinamik Görev Eşleme örneği

.expand()

@task
def fetch_files():
    return ["/data/file_1.csv", "/data/file_2.csv", "/data/file_3.csv"]

@task(max_active_tis_per_dagrun=2)  # azami paralelliği kontrol et
def process_file(path: str):
    print(f"processing file: {path}")


files = fetch_files()
process_file.expand(path=files)

Çalışma anında ölçekle

UI'da eşlenen görevler

Eşlenen örnekler tek bir görevin altında indeksli örnekler olarak görünür
Her birine tıklayıp tekil günlükler ve durumu gör
Yüzlerce örnek olsa bile Dag okunabilir kalır

UI'da eşlenen görevler

.partial()

@task
def process_file(path: str, output_dir: str):
    transform(path, output_dir)


files = get_files()
process_file.partial(output_dir="/out").expand(path=files)

Çarpraz-ürün sorunu

files = get_files()           # ["/data/a.csv", "/data/b.csv", "/data/c.csv"]
destinations = get_targets()  # ["s3://out/a", "s3://out/b", "s3://out/c"]


# Kartezyen çarpım: 3 x 3 = 9 örnek (istediğimiz bu değil!)
process.expand(path=files, dest=destinations)

Listeleri zip'le

.zip()

files = get_files()           # ["/data/a.csv", "/data/b.csv", "/data/c.csv"]
destinations = get_targets()  # ["s3://out/a", "s3://out/b", "s3://out/c"]


# Eşli: 3 örnek (a->a, b->b, c->c)
process.expand_kwargs(files.zip(destinations))

zip() öğeleri konuma göre bire bir eşler
expand_kwargs her çifti anahtar sözcük argümanlarına açar

Ne zaman hangisi

Desen	Kullanım
`expand()`	Tek bir liste üzerinde eşle
`partial()`	Örnekler arasında ortak parametreleri sabitle
`zip()`	Birden fazla listeyi konuma göre eşle

Ölçeklenebilir Dags oluşturmaya yardım eder
Veri değişince, görev sayısı çalışma anında otomatik ayarlanır

Hadi pratik yapalım!

Airflow ile Veri İş Hatları Oluşturma