Airflow Nasıl Kullanılır: Güvenilir Veri İşleme Hatları Oluşturmak İçin Uygulamalı, Uçtan Uca Bir Kılavuz
Veri taşıyor veya ML işlerini düzenliyorsanız, muhtemelen aynı nakaratı duymuşsunuzdur: "Sadece Airflow'a koy." Gerçek şu ki, Apache Airflow, karmaşık iş akışları üzerinde görünürlüğe, güvenilirliğe ve kontrole ihtiyaç duyduğunuzda parlar. Bu pratik kılavuzda, temel kavramlardan üretime hazır kalıplara kadar Airflow'u nasıl kullanacağınızı adım adım anlatacağız, böylece güvenebileceğiniz işlem hatları gönderebilirsiniz.
Uygulanabilir tutacağız: DAG'ler ve görevler için bir zihinsel model, TaskFlow API ile uygulamalı örnekler, dağıtım seçenekleri, test stratejileri ve en iyi uygulamalar elde edeceksiniz. Sonunda, "Eğitimi çalıştırabilirim"den "Bunu üretimde çalıştırabilirim"e geçeceksiniz.
Not: Daha derinlemesine incelemeler ve referans için, resmi belgeler mükemmel ve düzenli olarak güncellenmektedir.
Apache Airflow Gerçekte Nedir?
Airflow bir orkestratördür—bir veri işlemcisi değil. Başka bir yerde çalıştırdığınız işleri (veritabanları, veri ambarları, Spark işleri, API'ler, konteynerler) planlar, sıralar ve izler. İş akışlarını, görevleri ve bağımlılıklarını kodlayan Python dosyaları olan DAG'ler (Yönlendirilmiş Döngüsüz Grafikler) olarak tanımlarsınız. Airflow daha sonra bu görevleri programınıza, parametrelerinize ve ortamınıza göre yürütür.
- DAG: İş akışı tanımı (bağımlılıkları olan görevlerin grafiği).
- Görev: Bir iş birimi (Python fonksiyonu, SQL yürütme, Bash komutu, harici iş tetikleyicisi, vb.).
- Operatör: Bir tür görev için şablon (örn.,
PythonOperator, BashOperator, KubernetesPodOperator).
- Zamanlayıcı: Ne zaman ve neyin çalıştırılacağına karar verir.
- Yürütücü: Görevleri çalıştırır (yerel olarak, Celery ile, Kubernetes ile, vb.).
- UI: Çalıştırmalar, günlükler, yeniden denemeler ve soy için kontrol merkeziniz.
Airflow'u kurduktan sonra resmi eğitimlerle başlayın; size büyük resmi hızlı bir şekilde verirler.
Airflow'u Doğru Şekilde Kurma ve Çalıştırma
Airflow esnektir. Aşamanıza uygun yolu seçin:
- Yerel geliştirme (hızlı başlangıç):
- Proje tarafından sağlanan hızlı başlangıç Docker Compose'u kullanın. Web sunucusunu, zamanlayıcıyı, veritabanını ve daha fazlasını makul varsayılanlarla başlatır.
- DAG'leri öğrenmek ve yinelemek için harika.
- Küçük ekip veya hazırlık:
- Yönetilen bir Postgres ile Celery Executor veya Kubernetes Executor.
- Günlükleri S3/GCS'de saklayın ve bağımlılıkları resminizle veya
requirements.txt ile paketleyin.
- Esneklik için Kubernetes Executor veya otomatik ölçeklendirme çalışanlarına sahip Celery Executor.
- Harici sırlar (Vault), sağlam gözlemlenebilirlik (günlükler + metrikler) ve yükseltmeler için mavi/yeşil dağıtımlar.
İpucu: Airflow kod tabanınızı sürüm kontrollü, konteynerize edilmiş ve terfi etmeden önce test edilmiş tutun. "En İyi Uygulamalar" sayfası, üretime hazır kalıpları özetlemektedir.
Günlük Olarak Kullanacağınız Temel Kavramlar
DAG'ler: Kod Olarak İş Akışınız
Bir DAG, aşağıdakileri tanımlayan bir Python dosyasıdır:
- DAG meta verileri: id, zamanlama, başlangıç tarihi, etiketler.
- Varsayılan argümanlar: yeniden denemeler, sahipler, SLA'lar.
- Görevler ve bağımlılıkları.
Bir DAG'yi "ne" ve "ne zaman" olarak ve görevleri "nasıl" olarak düşünün.
Görevler ve Operatörler
Operatörler, yaygın görevler için hazır yapılardır. Örnekler:
- Python kodu için PythonOperator / TaskFlow
@task
- Kabuk komutları için BashOperator
- API'ler için SimpleHttpOperator
- Konteynerize edilmiş işler için KubernetesPodOperator
- Ambar işleri için SQL sağlayıcıları (örn., Snowflake, BigQuery, Postgres)
TaskFlow API: Modern, Pythonic Yol
TaskFlow API, görevleri @task ile Python fonksiyonları olarak yazmanıza, XCom aracılığıyla geçen değerleri döndürmenize ve bunları temiz bir şekilde oluşturmanıza olanak tanır. Boilerplate'i azaltır ve okunabilirliği artırır—şiddetle tavsiye edilir.
İlk Airflow DAG'niz (TaskFlow Sürümü)
Aşağıda, temel fikirleri göstermek için minimal bir ETL tarzı örnek bulunmaktadır: zamanlama, TaskFlow, bağımlılıklar ve XCom veri geçişi.
from datetime import datetime
from airflow import DAG
from airflow.decorators import task
with DAG(
dag_id="weather_etl_example",
start_date=datetime(2024, 1, 1),
schedule="0 * * * *", # hourly
catchup=False,
tags=.
Zamanlama, Catchup ve Backfill'ler
schedule: Cron veya ön ayarlar (@daily, @hourly).
start_date + catchup: Eğer catchup=True ise, Airflow başlangıç tarihinden itibaren çalıştırmaları geri dolduracaktır. Akış tarzı işlem hatları için catchup=False olarak ayarlayın.
- Manuel geri doldurmalar: Geçmiş aralıkları yeniden çalıştırmak için UI veya CLI'yı kullanın.
Pratik kural: deterministik toplu işler için catchup'ı etkinleştirin; gerçek zamanlı veya API hız sınırlı işlem hatları için devre dışı bırakın.
Görevler Arasında Veri Geçirme (XCom) Güvenli Bir Şekilde
- Küçük nesneler: TaskFlow ile dönüş değerleri iyidir.
- Büyük yükler: XCom'da bir anahtarla nesne depolamada (S3/GCS) saklayın.
- XCom'da hassas verilerden kaçının; sır arka uçlarını (örn., Vault) ve ortam değişkenlerini kullanın.
Dinamik Görev Eşleme ve Fan-out İş Yükleri
Airflow, girdilere göre çalışma zamanında dinamik olarak görevler oluşturabilir—bölümlenmiş veri kümeleri veya çok kiracılı işler için idealdir.
- DAG'leri deterministik ve idempotent tutun.
- Orkestrasyonu (Airflow) hesaplamadan (Spark, dbt, veri ambarları) ayırın.
- Açıklık ve XCom hijyeni için TaskFlow API'sini kullanın.
- DAG'leri parametreleştirin; değişkenleri dikkatli kullanın.
- İşlem hatlarınızı izleyin, uyarın ve belgeleyin.
Veri Ambarları ve ML ile Nasıl Çalışılır
- Veri ambarları: SQL işleri için sağlayıcı operatörlerini (örn., SnowflakeOperator, BigQueryInsertJobOperator) kullanın. SQL'i dosyalarda veya sürüm kontrollü modüllerde saklayın.
- dbt: Bash/KubernetesPodOperator veya sağlayıcılardaki özel dbt operatörleri aracılığıyla dbt'yi tetikleyin.
- ML: Özellik oluşturmayı, eğitimi ve toplu çıkarımı ayrı görevler olarak düzenleyin; artefaktları depolamada önbelleğe alın ve metrikleri günlüğe kaydedin.
Gelişmiş Zamanlama: Veri Kümeleri ve Çapraz DAG Bağımlılıkları
- Veri Kümeleri, bir DAG'nin, güncellendiğinde başka bir DAG'yi tetikleyen mantıksal bir veri kümesi üretmesine olanak tanır—ad-hoc tetikleyicilerden daha temizdir.
- Eski kalıplar için ExternalTaskSensor çalışır, ancak veri kümeleri daha deklaratiftir.
Güvenlik ve Uyumluluk
- UI'da rol tabanlı erişim kontrolü (RBAC) kullanın.
- Ortamları ekip veya güven sınırı başına yalıtın.
- Günlükler ve bağlantı değişikliği geçmişi aracılığıyla denetim izlerini saklayın.
Yükseltmeler ve Sürümleme
- Üretim benzeri iş yükleriyle hazırlıkta yükseltmeleri test edin.
- Sağlayıcıları kasıtlı olarak sabitleyin ve yükseltin.
- Yürütücüye özgü değişiklikler ve kullanımdan kaldırmalar için sürüm notlarını okuyun.
İlk Üretim DAG'niz İçin Hızlı Bir Kontrol Listesi
- Net sahiplik (
owner etiketi) ve uyarılar yapılandırıldı.
- Makul geri çekilmelerle
retries ayarlandı.
- Idempotent görevler ve açık bağımlılıklar.
- Küçük XCom yükleri; depolamada büyük veri.
- Dayanıklı depolamaya gönderilen günlükler; dışa aktarılan metrikler.
- Dağıtım planı (kanarya veya mavi/yeşil) ve geri alma adımları.
Örnek: Gerçekçi Bir Ambar Yükleme DAG'si
Bu kalıp, günlük dosyaları ayıklar, bunları doğrular ve bölüm başına dinamik eşleme ve ertelenebilir sensörlerle bir ambar tablosuna yükler.
from datetime import datetime
from airflow import DAG
from airflow.decorators import task
from airflow.sensors.filesystem import FileSensor
- Üretime terfi etmeden önce En İyi Uygulamaları gözden geçirin.
- Sistemleriniz (ambarlar, bulutlar, ML araçları) için sağlayıcı belgelerini keşfedin.
Bu arada: Bir AI yardımcısıyla yazmayı hızlandırın
Belirtmekte fayda var: çok sayıda DAG taslağı hazırlarsanız, kodu anlayan bir AI asistanı boilerplate'i hızlandırabilir, TaskFlow saplamaları oluşturabilir ve hatta bağımlılık düzeltmeleri önerebilir. Düzenleyicinizin ve tarayıcınızın yanında hafif bir yardımcıya ihtiyacınız varsa, Sider.AI geliştirme sırasında hızlı kod yeniden yazma ve açıklamalar için kullanışlı olabilir. Temel Çıkarımlar
- Hesaplama yapmak için değil, düzenlemek için Airflow'u kullanın.
- Temiz, test edilebilir DAG'ler için TaskFlow API'sini tercih edin.
- Verileri XCom'dan uzak tutun; bunun yerine referansları geçirin.
- Yuvaları kaydetmek için ertelenebilir sensörler/operatörler kullanın.
- Ortamlar aracılığıyla konteynerize edin, test edin ve terfi ettirin.
- Kuzey yıldızınız olarak resmi eğitimlere ve en iyi uygulamalara güvenin.
SSS
S1:Airflow'u nasıl kullanacağınızı öğrenmenin en kolay yolu nedir?
DAG'leri, görevleri, zamanlamayı ve UI'yı anlamak için resmi Eğitim ile başlayın. Ardından, küçük bir TaskFlow tabanlı işlem hattı oluşturun ve üretim hazırlığı için en iyi uygulamalar kılavuzuyla yineleyin.
S2:Airflow'da TaskFlow API'sini mi yoksa klasik operatörleri mi kullanmalıyım?
Çoğu Pythonic işlem hattı için TaskFlow API'sini kullanın, çünkü daha temizdir ve XCom dönüşlerini doğal olarak işler. Klasik operatörler, Bash, SQL veya konteyner işleri gibi Python olmayan görevler için hala harikadır.
S3:Airflow görevleri arasında büyük verileri nasıl geçiririm?
XCom'a büyük yükler koymaktan kaçının. Verileri S3/GCS'de veya bir veritabanında saklayın ve görevleri hızlı ve güvenilir tutmak için XCom aracılığıyla yalnızca referansları veya URI'leri geçirin.
S4:Üretimde Airflow için hangi yürütücüyü seçmeliyim?
Esneklik ve yalıtım için Kubernetes Executor güçlü bir varsayılandır. Daha basit kurulumlar için Celery Executor iyi çalışır—sadece otomatik ölçeklendirme, sağlam günlük kaydı ve haricileştirilmiş sırları sağlayın.
S5:Birden çok Airflow DAG'si arasında bağımlılıkları nasıl ele alırım?
Bir işlem hattı başka biri için veri ürettiğinde, deklaratif çapraz DAG tetikleyicileri için Veri Kümelerini kullanın. Alternatif olarak, ExternalTaskSensor çalıştırmaları koordine edebilir, ancak Veri Kümeleri veri odaklı orkestrasyon için daha temizdir.
