Technical Tutorials

Modern veri dünyasında karmaşık veri iş akışlarını güvenilir bir şekilde orkestra etme yeteneği sadece bir kolaylık değil, aynı zamanda iş açısından kritik bir gereksinimdir. Apache Airflow, programlı olarak iş akışları oluşturma, zamanlama ve izleme konusunda varsayılan standart haline gelmiştir. Hattları kod olarak ele alarak Airflow, veri mühendislerinin tekrarlanabilir, sürüm kontrollü ve sağlam veri altyapıları oluşturmalarını sağlar. Bu yazıda, temel kavramlardan ileri düzey üretim dağıtım stratejilerine kadar Airflow'un temel mekaniklerini keşfedeceğiz.

Temel Soyutlama: DAG'ler ve Zamanlama

Her Airflow örneğinin kalbinde Yönlendirilmiş Döngüsüz Grafik (DAG) bulunur. Bir DAG, iş akışınızı görevler ve bunların bağımlılıkları koleksiyonu olarak tanımlar. "Yönlendirilmiş" yönü, döngüsel bağımlılıkların olmamasını sağlarken; "Döngüsüz" yönü, her görevin bir başlangıç noktasından ulaşılabileceğini garanti eder. Airflow'ta zamanlama zaman tabanlıdır. Bir zamanlama aralığı (cron benzeri sözdizimi) tanımlarsınız ve zamanlayıcı, her DAG çalıştırmasının ne zaman tetiklenmesi gerektiğini belirler. Bu yürütme zamanının tanımlama zamanından ayrıştırılması, idempotent (tekrarlanabilir) hat çalıştırmalarına olanak tanır. Örneğin, geçici bir ağ hatası nedeniyle bir hatta başarısız olursa, aynı mantıksal aralığı tekrar çalıştırmak doğru tasarlanmışsa veriyi çoğaltmaz.

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

default_args = {
    'owner': 'data_engineer',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'retries': 3,
    'retry_delay': timedelta(minutes=5)
}

with DAG(
    'etl_daily_summary',
    default_args=default_args,
    description='Günlük satış verileri için ETL hattı',
    schedule_interval='@daily',
    catchup=False
) as dag:
    def extract_data(**kwargs):
        print("Ham satış verileri çıkarılıyor...")

    extract_task = PythonOperator(
        task_id='extract',
        python_callable=extract_data
    )

Operatörler, Sensörler ve Görev Bağımlılıkları

DAG içindeki görevler, Operatörler olarak adlandırılan farklı bileşenler tarafından yürütülür. PythonOperator en esnek olanı olsa da, Airflow SQL (PostgresOperator, MySqlOperator), Bulut Depolama (S3Hook) ve Kubernetes (KubernetesPodOperator) için özel operatörler sağlar. Sabit bir zaman veya yukarı akış görevi tamamlamasından ziyade harici koşullara dayanan bağımlılıklar için Sensörler kullanılır. Bir sensör, S3 kovasına bir dosyanın gelmesi veya bir veritabanı kaydının güncellenmesi gibi belirli bir koşulun yerine getirilmesini bekler. Bu, olay tabanlı veri mimarileri için hayati önem taşır.

from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor

wait_for_data = S3KeySensor(
    task_id='wait_for_new_file',
    bucket_key='data/incoming/sales_*.csv',
    bucket_name='my-data-lake',
    timeout=3600,
    poke_interval=60
)

ETL Otomasyonu ve En İyi Uygulamalar

Airflow'ta bir ETL hattı oluşturmak, sorumlulukların sıkı bir şekilde ayrılmasını gerektirir. Çıkarma, Dönüştürme ve Yükleme aşamaları, ince taneli yeniden deneme mantığı ve izleme için genellikle ayrı görevler olmalıdır. Her şeyi yapan monolitik Python betikleri yazmaktan kaçının. Bunun yerine, dönüşümleri daha küçük, test edilebilir işlevlere ayırın. Ayrıca, dinamik değerleri görevler arasında aktarmak için Airflow'un şablonlama motorunu (Jinja) kullanın. Bu, kod tekrarını azaltır ve DAG'lerinizi daha sürdürülebilir hale getirir. Örneğin, mevcut yürütme tarihine göre dosya yollarını veya SQL sorgularını dinamik olarak oluşturabilirsiniz.

Üretim Dağıtımları

Airflow'u üretim ortamında çalıştırmak, dikkatli mimari planlama gerektirir. Varsayılan SQLite veritabanı geliştirme için uygun olsa da, üretim ortamları meta veri depolama için PostgreSQL veya MySQL gibi sağlam bir ilişkisel veritabanı kullanmalıdır. Yüksek kullanılabilirlik, birden fazla Airflow zamanlayıcısı ve işçisi çalıştırılarak sağlanır. Ölçeklenebilirlik için, her görev için dinamik olarak Kubernetes pod'ları oluşturan KubernetesExecutor'ı kullanmayı düşünün; bu sayede bulut-native kaynak yönetiminin avantajlarından yararlanabilirsiniz. Docker imajlarınızın hafif ve tekrarlanabilir olduğundan emin olun ve dağıtımdan önce DAG mantığını test etmek için CI/CD hatları kullanın.

Sonuç

Apache Airflow, iş akışı orkestrasyonu için güçlü, kod odaklı bir yaklaşım sunar. DAG'leri ustalaşarak, operatörler ve sensörlerin nüanslarını anlayarak ve üretim seviyesinde dağıtım uygulamalarına uyarak, yalnızca işlevsel değil, aynı zamanda dayanıklı, ölçeklenebilir ve sürdürülebilir veri hatları oluşturabilirsiniz. Veri karmaşıklığı arttıkça Airflow, veri mühendisinin silah deposunda vazgeçilmez bir araç olarak kalmaya devam ediyor.