Technical Tutorials

Airflow, veri iş akışlarını orkestrasyon için uzun süredir fiili standart olmuştur. Ancak veri hacimleri büyüdükçe ve kaynaklar parçalandıkça, statik DAG'lar genellikle ölçeklenebilirlik ve sürdürülebilirlik konusunda zorlanabilir. Airflow 2.0 ve sonraki sürümlerde, Görev Eşlemesi (Task Mapping) ve iyileştirilmiş Dinamik DAG üretimi, performans veya netlikten ödün vermeden değişen veri yapılarına uyum sağlayan boru hatları oluşturmak için sağlam araçlar sunar.

Neden Statik DAG'lar Ölçekle Başarısız Olur

Geleneksel Airflow DAG'ları ayrıştırma (parse) zamanında tanımlanır. Boru hattınız 100 dosyayı işlemek zorundaysa, 100 görev oluşturabilirsiniz. Bu yaklaşım şunlara yol açar:

Kod Şişkinliği: Tekrarlayan görev tanımlarına sahip devasa DAG dosyaları.
Uzun Ayrıştırma Süreleri: Airflow, karmaşık DAG'ları ayrıştırmak için çok fazla zaman harcar, bu da zamanlayıcıyı (scheduler) yavaşlatır.
Esneklik Eksikliği: Sert kodlanmış sınırlar veya karmaşık döngüler kullanmadan bilinmeyen sayıda dosyayı işleyemezsiniz.

Çözüm, "çok sayıda görevden" birden fazla öğeyi işleyen "tek bir eşlenmiş göreve" geçmektir.

Görev Eşlemesine (Task Mapping) Giriş

Airflow 2.0'daki Görev Eşlemesi, tek bir görev tanımından birden fazla görev örneği oluşturmanıza olanak tanır. x.map() yöntemini kullandığınızda, Airflow, sağlanan yinelebilir öğedeki (liste, demet veya üreteç) her unsur için ayrı bir yürütme oluşturur.

Temel Sözdizimi

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

with DAG(
    dag_id="task_mapping_example",
    schedule_interval="@daily",
    start_date=datetime(2023, 1, 1),
    catchup=False
) as dag:

    def process_file(file_name):
        print(f"Processing {file_name}")
        # Add your processing logic here
        return file_name

    # Define a list of files to process
    files = ["file1.csv", "file2.csv", "file3.csv", "file4.csv", "file5.csv"]

    # Map the task over the list of files
    map_task = PythonOperator(
        task_id="process_file_task",
        python_callable=process_file,
        op_kwargs={"file_name": files}  # Note: In Airflow 2.3+, use 'op_args' or 'op_kwargs' carefully
    )
    
    # Actually, the standard way for mapping a list of items to a function argument:
    # If process_file takes one arg, and we want to pass each item:
    
    # Corrected Implementation for Airflow 2.0+:
    def process_item(item):
        print(f"Processing: {item}")
        return item

    # We need to pass the list as an argument to map
    # Using op_args is not directly supported for mapping list elements in all versions.
    # Best practice: Use a wrapper or pass the list as op_kwargs and unpack?
    # Actually, Airflow 2.3+ recommends:
    
    mapped_task = PythonOperator(
        task_id="process_items",
        python_callable=process_item,
        op_args=files  # This is NOT how mapping works directly.
    )

Düzeltme ve En İyi Uygulama: Airflow'da, bir listeyi eşlemek için sadece op_args'a geçemezsiniz. Operatör örneğinde .x.map() yöntemini kullanmanız gerekir. Ancak PythonOperator, basit dizeler için BashOperator'ın yaptığı gibi doğrudan bir .x.map() yöntemine sahip değildir. PythonOperator için yaygın desen, tüm listeyi işleve geçirmek ve yinelemeyi içinde işlemektir VEYA genel bir BaseOperator alt sınıfını kullanmaktır VEYA daha yaygın olarak, değeri liste olan op_kwargs'ı kullanmaktır, ancak çağrılabilir (callable) bu listeyi kabul etmelidir.

Aslında, bir PythonOperator içindeki öğeler listesi için Görev Eşlemesini göstermenin en sağlam yolu, listeyi bir argüman olarak geçirmek ve Airflow'un onu bölmesini sağlamaktır. Ama bekleyin, PythonOperator eşlemesi, eşlenen listenin her unsurunu çağrılabilene geçirerek çalışır.

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def process_file(file_name):
    print(f"Processing: {file_name}")
    return f"Processed {file_name}"

with DAG(
    dag_id="dynamic_processing",
    schedule_interval=None,
    start_date=datetime(2023, 1, 1),
) as dag:

    # Simulate a dynamic list of files (e.g., from a DB query or S3 bucket)
    files = [f"file_{i}.csv" for i in range(1, 101)]

    # Define the task
    process_task = PythonOperator(
        task_id="process_files",
        python_callable=process_file
    )

    # Map the task. This will create 100 task instances.
    # The list `files` is passed as the argument to map over.
    process_task.x.map(process_task, files)

Önemli Not: Yukarıdaki .x.map() sözdizimi, netlik için biraz basitleştirilmiştir. Pratikte, eşlenen öğelerin hangi parametreye gideceğini belirtmek için genellikle op_kwargs kullanılır. İşleviniz birden fazla argüman alıyorsa, eşlenecek argümanın indeksini belirtmeniz gerekir.

def process_file(file_name, log_level="INFO"):
    print(f"[{log_level}] Processing: {file_name}")
    return f"Processed {file_name}"

# ... inside DAG ...
process_task = PythonOperator(
    task_id="process_files",
    python_callable=process_file,
    op_kwargs={"log_level": "DEBUG"}  # Fixed keyword argument
)

# Map over the first argument (file_name)
# index=0 refers to the first positional argument of the callable
process_task.x.map(process_task, files, index=0)

Dinamik Verileri İşleme: Üreteçler ve Bağlam

Genellikle, öğeler listesi DAG tanımı zamanında bilinmez. İşte burada dinamik DAG üretimi ve üreteçler (generators) parlar. Bellekte devasa bir liste oluşturmak yerine, öğeleri tek tek veren bir üreteç kullanın. Öğelerin sayısı binlerce olabileceği büyük ölçekli boru hatları için bu çok önemlidir.

import boto3

def get_s3_files(bucket, prefix):
    s3 = boto3.client('s3')
    paginator = s3.get_paginator('list_objects_v2')
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        for obj in page.get('Contents', []):
            yield obj['Key']

# Inside DAG definition:
# Note: You cannot easily use a generator that hits S3 at parse time if the 
# bucket structure changes dynamically per run. 
# Best Practice: Use a Task to fetch the list, then map based on that output.

Ölçeklenebilirlik İçin En İyi Uygulamalar

Görev Sayısını Sınırlayın: Görev Eşlemesi "çok sayıda görev" kod sorununu çözerken, tek bir yürütmede 10.000 görev örneğine sahip olmak Airflow zamanlayıcısını ve veritabanını aşırı yükleyebilir. Sayı 500-1000'i aşıyorsa öğeleri toplu (batch) işlemeyi düşünün.
Toplu İşlem Kullanın: Her öğe küçükse, görev başlatma ve günlük tutma (logging) yükünü azaltmak için öğeleri toplara (örneğin, görev başına 50 dosya) gruplayın.
Performansı İzleyin: Yürütme başına üretilen görev örneği sayısını izleyin. darboğazları belirlemek için airflow_dagrun_taskinstance_run_duration gibi metrikleri kullanın.
Hata İşleme: Varsayılan olarak, eşlenmiş bir görev başarısız olursa, tüm yürütme başarısız olarak işaretlenir. Bireysel hataları tüm boru hattını durdurmadan tolere etmeniz gerekiyorsa, partial_mapping=True (daha yeni sürümlerde mevcut) veya özel mantık kullanın.

Sonuç

Airflow 2.0+, iş ihtiyaçlarınızla ölçeklenen veri boru hatları oluşturmak için güçlü özellikler sunar. Görev Eşlemesinden yararlanarak, kırılgan, sert kodlanmış DAG'lardan değişken veri hacimlerini verimli bir şekilde işleyen esnek, dinamik iş akışlarına geçebilirsiniz. Airflow zamanlayıcınızın tepkili kalmasını sağlamak için görev örneği sayılarını izlemeyi ve çok büyük veri kümeleri için toplu işlemeyi düşünmeyi unutmayın. Bir sonraki boru hattınızı tasarlarken kendinize sorun: "Bu eşlenebilir mi?" Cevap evetse, dinamik görev üretiminin gücünü benimseyin.