Python’dan Üretim Hattına: Apache Airflow ile Pratik Bir Kılavuz

Python’dan Üretim Hattına: Apache Airflow ile Pratik Bir Kılavuz

Günümüz veri odaklı dünyasında, şirketler her geçen gün daha fazla veri üretmekte ve bu veriyi anlamlı içgörülere dönüştürmek için karmaşık süreçlere ihtiyaç duymaktadır. Manuel olarak yönetilen veri işleme adımları, zamanla bir kabusa dönüşebilir: hatalara açık, ölçeklenemez ve izlenmesi zor hale gelirler. Peki, yüzlerce farklı kaynaktan gelen veriyi düzenli, hatasız ve otomatik bir şekilde işleyip, doğru zamanda doğru yere ulaştıracak sağlam bir üretim hattını (production pipeline) nasıl kurabiliriz? İşte tam bu noktada, Python’ın gücüyle entegre çalışan Apache Airflow, veri mühendislerinin ve geliştiricilerin en büyük yardımcısı olarak sahneye çıkıyor. Bu kapsamlı rehberde, Python projelerinizi üretim ortamına taşımak için Apache Airflow’u nasıl kullanacağınızı adım adım keşfedeceğiz.

Apache Airflow Nedir ve Veri Akış Yönetiminde Neden Bu Kadar Önemlidir?

Apache Airflow, programatik olarak iş akışlarını (workflows) yazmanıza, planlamanıza ve izlemenize olanak tanıyan açık kaynaklı bir platformdur. Özellikle veri mühendisliği, makine öğrenimi ve iş zekası alanlarında, birbirine bağımlı görevlerin (tasks) karmaşık zincirlerini otomatikleştirmek için vazgeçilmez bir araç haline gelmiştir. Python ile yazılmış olması, platformu Python ekosistemine aşina olan geliştiriciler için son derece erişilebilir kılar.

Peki, Airflow’u bu kadar önemli yapan nedir?

• Programatik İş Akışı Tanımlama (Programmatic Workflow Definition): Airflow’da iş akışları, yani DAG’ler (Directed Acyclic Graphs – Yönlendirilmiş Döngüsel Olmayan Grafikler), saf Python kodu ile tanımlanır. Bu, sürüm kontrol sistemleriyle (örneğin Git) kolayca entegre olabilmesini, test edilebilirliğini ve esnekliğini artırır. Bir veri hattını (data pipeline) kod olarak düşünmek, “altyapı kod olarak” (infrastructure as code) prensibinin iş akışlarına uygulanması anlamına gelir.
• Ölçeklenebilirlik (Scalability): Airflow, dağıtık bir mimariye sahiptir. Görevleri birden fazla worker (işçi) üzerinde paralel olarak çalıştırabilir, bu da büyük veri kümeleriyle çalışırken veya çok sayıda iş akışı yönetirken performansı artırır. Kubernetes veya Celery gibi teknolojilerle entegrasyonu sayesinde yatay ölçeklenebilirlik sunar.
• Zengin Kullanıcı Arayüzü (Rich User Interface): Airflow’un web tabanlı kullanıcı arayüzü, DAG’lerinizi görselleştirmek, görevlerin durumunu izlemek, hata günlüklerini (logs) incelemek ve iş akışlarını manuel olarak tetiklemek veya durdurmak için merkezi bir kontrol paneli sağlar. Bu, operasyonel görünürlüğü (operational visibility) önemli ölçüde artırır.
• Esneklik ve Genişletilebilirlik (Flexibility and Extensibility): Airflow, çeşitli veri kaynakları ve platformlarla entegrasyonu kolaylaştıran yüzlerce operatör (operators), sensör (sensors) ve hook (kancalar) ile birlikte gelir. Örneğin, bir veritabanından veri çekmek, bir API’ye istek göndermek, Spark işleri çalıştırmak veya bir bulut depolama hizmetine dosya yüklemek gibi görevler için önceden tanımlanmış operatörler mevcuttur. Kendi özel operatörlerinizi ve hook’larınızı yazarak Airflow’u kendi özel ihtiyaçlarınıza göre genişletebilirsiniz.
• Hata Yönetimi ve Yeniden Deneme Mekanizmaları (Error Handling and Retries): Bir üretim ortamında hatalar kaçınılmazdır. Airflow, görevlerin başarısız olması durumunda otomatik yeniden deneme (retries) mekanizmaları, hata bildirimleri (e-posta, Slack vb.) ve esnek hata işleme stratejileri sunar. Bu özellikler, veri hatlarınızın sağlamlığını ve güvenilirliğini artırır.

Özetle, Airflow, karmaşık veri işleme süreçlerini otomatikleştirmek, yönetmek ve izlemek için güçlü, esnek ve ölçeklenebilir bir çözüm sunar. Bir veri mühendisi veya geliştirici olarak, zamanınızı tekrarlayan manuel görevler yerine daha değerli işlere ayırmanıza olanak tanır.

Airflow’un Temel Bileşenleri Nelerdir ve Mimari Nasıl Çalışır?

Apache Airflow, bir dizi temel bileşenin uyumlu bir şekilde çalışmasıyla tam işlevselliğini sunar. Bu bileşenlerin her biri, veri iş akışlarınızın sorunsuz bir şekilde yürütülmesini sağlamak için belirli bir role sahiptir. Airflow’un mimarisini anlamak, sorun giderme ve optimizasyon için kritik öneme sahiptir.

Airflow’un başlıca bileşenleri şunlardır:

1. Web Sunucusu (Webserver): Airflow’un kullanıcı arayüzünü (UI) barındıran bileşendir. Geliştiricilerin ve operasyon ekiplerinin DAG’leri görselleştirmesine, görevlerin durumunu izlemesine, günlükleri görüntülemesine, bağlantıları ve değişkenleri yönetmesine olanak tanır. Python’da Flask framework’ü ile oluşturulmuştur ve genellikle bir HTTP sunucusu (örneğin Gunicorn) aracılığıyla çalışır.
2. Planlayıcı (Scheduler): Airflow’un kalbi diyebiliriz. Bu bileşen, DAG klasörünüzü sürekli olarak izler, yeni DAG’leri veya güncellemeleri algılar ve görevlerin zamanlanmış tetikleyicilerine göre çalıştırılması gereken görevleri belirler. Planlayıcı, görev bağımlılıklarını (dependencies) çözer, görevlerin çalışmaya hazır olup olmadığını kontrol eder ve uygun görevleri yürütmek üzere bir yürütücüye (Executor) gönderir. Airflow’un zamanlama mantığı, cron benzeri ifadelerden veya özel zamanlama ayarlarından beslenir.
3. Yürütücü (Executor): Görevlerin (tasks) gerçekten nerede ve nasıl çalıştırılacağını belirleyen bileşendir. Airflow’un esnekliği burada da kendini gösterir, çünkü farklı yürütücü türleri mevcuttur:
   • SequentialExecutor: Geliştirme ortamları için idealdir. Tüm görevleri tek bir işlemde, sırayla çalıştırır. Üretim için uygun değildir.
   • LocalExecutor: Görevleri aynı makinedeki paralel süreçlerde çalıştırır. Daha büyük geliştirme veya küçük üretim ortamları için kullanılabilir, ancak tek bir makinenin kaynaklarıyla sınırlıdır.
   • CeleryExecutor: Görevleri Celery kuyruğu aracılığıyla birden fazla worker (işçi) arasında dağıtır. Bu, Airflow kurulumunuzu yatay olarak ölçeklendirmenizi sağlar ve üretim ortamları için popüler bir seçenektir.
   • KubernetesExecutor: Her bir Airflow görevini ayrı bir Kubernetes Pod’u olarak çalıştırır. Bu, görevlerin birbirinden izole olmasını, kaynak izolasyonu sağlamasını ve Kubernetes’in güçlü orkestrasyon yeteneklerinden faydalanmasını sağlar. Bulut tabanlı üretim ortamları için en modern ve ölçeklenebilir çözümlerden biridir.

   Yürütücü seçimi, Airflow kurulumunuzun ölçeklenebilirlik ve esneklik ihtiyaçlarına göre yapılır.

4. Veritabanı (Metadata Database): Airflow’un tüm meta verilerini (meta-data) depoladığı yerdir. Bu veritabanı, DAG’lerin durumunu, görev örneklerinin (task instances) geçmişini, bağlantı bilgilerini, değişkenleri, kullanıcıları ve diğer yapılandırma verilerini içerir. Genellikle PostgreSQL veya MySQL gibi bir ilişkisel veritabanı kullanılır. Web sunucusu, planlayıcı ve yürütücüler, bu veritabanı aracılığıyla birbirleriyle iletişim kurar.
5. DAG Klasörü (DAGs Folder): Airflow’un DAG tanımlamalarını içeren Python scriptlerini bulduğu dosya sistemindeki bir dizindir. Planlayıcı, bu klasörü belirli aralıklarla tarayarak yeni veya güncellenmiş DAG’leri tespit eder ve veritabanına kaydeder.

Bu bileşenler, bir araya gelerek Airflow’un güçlü ve esnek bir iş akışı yönetim platformu olmasını sağlar. Planlayıcı, veritabanına bakarak hangi görevlerin çalıştırılması gerektiğini belirler, yürütücü bu görevleri uygun worker’lara dağıtır ve web sunucusu da tüm bu sürecin görsel bir temsilini sunar. Veritabanı ise tüm bu işlemlerin kaydını tutar ve bileşenler arası iletişimi mümkün kılar.

Airflow Ortamı Nasıl Kurulur ve Yapılandırılır?

Apache Airflow’u kurmanın ve yapılandırmanın birkaç yolu vardır, ancak geliştirme ve test ortamları için en yaygın ve önerilen yöntemlerden biri Docker Compose kullanmaktır. Docker Compose, Airflow’un tüm bileşenlerini (web sunucusu, planlayıcı, veritabanı vb.) tek bir komutla ayağa kaldırmanıza olanak tanır. Üretim ortamları için ise genellikle Kubernetes veya daha karmaşık Docker Swarm kurulumları tercih edilir.

Bu bölümde, Docker Compose kullanarak yerel bir Airflow ortamını nasıl kuracağınızı adım adım inceleyeceğiz.

Adım 1: Ön Koşullar

Öncelikle sisteminizde Docker ve Docker Compose’un kurulu olduğundan emin olun. Eğer kurulu değilse, işletim sisteminize uygun kurulum adımlarını Docker’ın resmi web sitesinden takip edebilirsiniz.

Adım 2: Airflow Docker Compose Dosyasını İndirme

Airflow’un resmi GitHub deposunda, hazır bir Docker Compose dosyası bulunur. Bu dosyayı indirmek için aşağıdaki komutları kullanabilirsiniz:

mkdir airflow_project
cd airflow_project
curl -LfO "https://airflow.apache.org/docs/apache-airflow/2.7.3/docker-compose.yaml"
  

Bu komutlar, airflow_project adında yeni bir dizin oluşturur ve Airflow’un Docker Compose dosyasını bu dizine indirir. İndirdiğiniz dosyanın adı docker-compose.yaml olacaktır. Airflow’un sürümüne göre URL’yi güncellemeyi unutmayın, yukarıdaki örnek 2.7.3 sürümü içindir.

Adım 3: Ortam Değişkenlerini Ayarlama

Docker Compose dosyasını çalıştırmadan önce, Airflow’un kullanıcı kimliği ve parolasını ayarlamak için bir .env dosyası oluşturmanız önerilir. Aynı dizinde .env adında bir dosya oluşturun ve içine aşağıdaki satırları ekleyin:

_AIRFLOW_WWW_USER_USERNAME=airflow
_AIRFLOW_WWW_USER_PASSWORD=airflow
  

Bu, varsayılan kullanıcı adını ve parolasını airflow olarak ayarlar. Üretim ortamında kesinlikle daha güvenli parolalar kullanmalısınız.

Adım 4: Airflow Ortamını Başlatma

Artık Airflow bileşenlerini başlatmaya hazırsınız. İlk çalıştırmada, Airflow’un gerekli dizinlerini oluşturması ve veritabanını başlatması için bir başlangıç komutu çalıştırmanız gerekir:

docker compose up airflow-init
  

Bu komut tamamlandıktan sonra, tüm Airflow hizmetlerini başlatabilirsiniz:

docker compose up -d
  

-d bayrağı, hizmetleri arka planda (detached mode) çalıştırmanızı sağlar.

Adım 5: Airflow UI’ya Erişme

Hizmetler başladıktan sonra, web tarayıcınızı açın ve http://localhost:8080 adresine gidin. Karşınıza Airflow’un giriş ekranı çıkacaktır. Adım 3’te ayarladığınız kullanıcı adı (airflow) ve parola (airflow) ile giriş yapabilirsiniz.

Adım 6: DAG Klasörünü Yapılandırma

Airflow’un DAG’lerinizi bulabilmesi için, Python scriptlerinizi içeren bir klasör oluşturmanız ve bu klasörü Docker Compose dosyasında Airflow konteynerine bağlamanız (mount) gerekir.
airflow_project dizininin içinde dags adında bir klasör oluşturun:

mkdir dags
  

Şimdi docker-compose.yaml dosyasını açın ve airflow-worker, airflow-scheduler ve airflow-webserver servislerinin volumes bölümüne aşağıdaki satırı ekleyin:

    volumes:
      - ./dags:/opt/airflow/dags
  

Bu değişiklikten sonra Docker Compose hizmetlerini yeniden başlatmanız gerekebilir:

docker compose down
docker compose up -d
  

Artık dags klasörünüze koyduğunuz Python dosyaları, Airflow UI’da DAG olarak görünecektir. Bu kurulum, yerel geliştirme ve test için sağlam bir temel sağlar. Üretim ortamları için ise, güvenlik, yedeklilik ve izleme gibi ek yapılandırmalar gerekecektir.

İlk DAG’ımızı Nasıl Oluştururuz? (Uygulamalı Örnek)

Airflow’un kalbi olan DAG’ler (Directed Acyclic Graphs – Yönlendirilmiş Döngüsel Olmayan Grafikler), iş akışlarınızı tanımladığınız Python dosyalarıdır. Bir DAG, bir dizi görevin (tasks) ve bu görevler arasındaki bağımlılıkların (dependencies) mantıksal bir koleksiyonudur. Her görev, bir operatör (operator) tarafından tanımlanır ve belirli bir işi yapar. Şimdi, ilk basit DAG’ımızı oluşturalım ve temel bileşenlerini inceleyelim.

Yukarıdaki bölümde oluşturduğumuz airflow_project/dags klasörünün içine ilk_dag.py adında bir dosya oluşturalım ve aşağıdaki kodu yapıştıralım:

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

# Python fonksiyonu tanımlayalım
def merhaba_dunya_fonksiyonu():
    print("Merhaba Airflow dünyası!")
    return "Fonksiyon başarıyla çalıştı."

def veri_cek_fonksiyonu(kaynak, **kwargs):
    print(f"{kaynak} kaynağından veri çekiliyor...")
    # Burada gerçek bir API çağrısı veya veritabanı sorgusu olabilir
    cekilen_veri = f"'{kaynak}' kaynağından çekilen örnek veri."
    print(f"Çekilen veri: {cekilen_veri}")
    # XCom kullanarak veriyi diğer görevlere aktarabiliriz
    kwargs['ti'].xcom_push(key='cekilen_veri', value=cekilen_veri)

def veriyi_isle_fonksiyonu(**kwargs):
    ti = kwargs['ti']
    cekilen_veri = ti.xcom_pull(key='cekilen_veri', task_ids='veri_cek_gorevi')
    print(f"İşlenecek veri: {cekilen_veri}")
    islenmis_veri = f"'{cekilen_veri}' işlendi ve dönüştürüldü."
    print(f"İşlenmiş veri: {islenmis_veri}")
    kwargs['ti'].xcom_push(key='islenmis_veri', value=islenmis_veri)

def veriyi_yukle_fonksiyonu(**kwargs):
    ti = kwargs['ti']
    islenmis_veri = ti.xcom_pull(key='islenmis_veri', task_ids='veriyi_isle_gorevi')
    print(f"Yüklenecek veri: {islenmis_veri}")
    print(f"'{islenmis_veri}' veritabanına veya depolama alanına yüklendi.")


with DAG(
    dag_id='ilk_airflow_dag',
    start_date=datetime(2023, 1, 1),
    schedule_interval='@daily', # Her gün çalışacak
    catchup=False, # Geçmişteki kaçırılan çalıştırmaları yapma
    tags=['ornek', 'python', 'egitim'],
    description='Basit bir Airflow DAG örneği',
) as dag:
    # Görev 1: Bash komutu çalıştırma
    bash_selamlama = BashOperator(
        task_id='bash_selamlama_gorevi',
        bash_command='echo "Merhaba, Airflow Bash Operatörü!"',
    )

    # Görev 2: Python fonksiyonu çalıştırma
    python_selamlama = PythonOperator(
        task_id='python_selamlama_gorevi',
        python_callable=merhaba_dunya_fonksiyonu,
    )

    # Görev 3: Veri çekme simülasyonu
    veri_cek_gorevi = PythonOperator(
        task_id='veri_cek_gorevi',
        python_callable=veri_cek_fonksiyonu,
        op_kwargs={'kaynak': 'API_Servisi'},
    )

    # Görev 4: Veriyi işleme simülasyonu
    veriyi_isle_gorevi = PythonOperator(
        task_id='veriyi_isle_gorevi',
        python_callable=veriyi_isle_fonksiyonu,
    )

    # Görev 5: Veriyi yükleme simülasyonu
    veriyi_yukle_gorevi = PythonOperator(
        task_id='veriyi_yukle_gorevi',
        python_callable=veriyi_yukle_fonksiyonu,
    )

    # Görevler arası bağımlılıkları tanımlama
    # bash_selamlama -> python_selamlama
    # python_selamlama -> veri_cek_gorevi
    # veri_cek_gorevi -> veriyi_isle_gorevi -> veriyi_yukle_gorevi

    bash_selamlama >> python_selamlama >> veri_cek_gorevi >> veriyi_isle_gorevi >> veriyi_yukle_gorevi
  

Kod Açıklaması:

1. İçe Aktarmalar (Imports):
   • DAG: Airflow’da bir iş akışını tanımlamak için ana sınıftır.
   • BashOperator: Bash komutlarını çalıştırmak için kullanılır.
   • PythonOperator: Python fonksiyonlarını çalıştırmak için kullanılır.
   • datetime: Zamanlama için kullanılır.
2. Python Fonksiyonları:
   • merhaba_dunya_fonksiyonu(): Basit bir mesaj yazdıran fonksiyon.
   • veri_cek_fonksiyonu(): Belirli bir kaynaktan veri çekme işlemini simüle eder ve çekilen veriyi XCom (Cross-Communication) kullanarak diğer görevlere aktarır. op_kwargs ile operatöre argüman geçirebiliriz.
   • veriyi_isle_fonksiyonu(): veri_cek_gorevi‘nden çekilen veriyi alır, işler ve işlenmiş veriyi yine XCom ile aktarır.
   • veriyi_yukle_fonksiyonu(): veriyi_isle_gorevi‘nden işlenmiş veriyi alır ve yükleme işlemini simüle eder.
3. DAG Tanımlaması:
   • dag_id: DAG’in benzersiz tanımlayıcısıdır. Airflow UI’da bu isimle görünür.
   • start_date: DAG’in ne zaman çalışmaya başlayacağını belirtir. Airflow, bu tarihten itibaren schedule_interval‘e göre geçmiş çalıştırmaları planlar.
   • schedule_interval: DAG’in ne sıklıkta çalışacağını belirler. Cron ifadeleri (örneğin '0 0 * * *' veya '@daily') veya zaman aralıkları (örneğin timedelta(days=1)) kullanılabilir. None olarak ayarlanırsa sadece manuel olarak tetiklenebilir.
   • catchup=False: start_date ile mevcut tarih arasındaki kaçırılan çalıştırmaların yapılmamasını sağlar. Genellikle geliştirme ortamlarında False olarak ayarlanır.
   • tags: DAG’leri gruplandırmak ve filtrelemek için etiketler.
   • description: DAG hakkında kısa bir açıklama.
4. Görev Tanımlamaları (Task Definitions):
   • BashOperator: task_id ile benzersiz bir kimlik ve bash_command ile çalıştırılacak Bash komutunu alır.
   • PythonOperator: task_id ve çalıştırılacak Python fonksiyonunu (python_callable) alır. Fonksiyona argüman geçirmek için op_kwargs kullanılabilir. XCom’a erişim için fonksiyonun **kwargs alması ve ti = kwargs['ti'] ile TaskInstance objesine ulaşması gerekir.
5. Bağımlılıklar (Dependencies):
   • Görevler arasındaki yürütme sırasını belirler. >> operatörü “şundan sonra çalış” anlamına gelir. Örneğin, task_a >> task_b, task_a başarıyla tamamlandıktan sonra task_b‘nin çalışacağı anlamına gelir. Birden fazla bağımlılık zincirleme yapılabilir.

Bu dosyayı airflow_project/dags klasörüne kaydettikten sonra, Airflow web arayüzünü yenilediğinizde “ilk_airflow_dag” adında yeni bir DAG göreceksiniz. DAG’i etkinleştirmek için anahtar düğmesini “On” konumuna getirin. Daha sonra, DAG’i manuel olarak tetikleyebilir veya zamanlanmış çalıştırmasını bekleyebilirsiniz. Çalıştırma sırasında görevlerin durumunu (running, success, failed) ve günlüklerini (logs) web arayüzünden izleyebilirsiniz. Bu basit örnek, Airflow’un temel gücünü ve esnekliğini göstermektedir.

Gerçek Dünya Senaryolarında Airflow Kullanımı: Bir Vaka Analizi

Birçok şirket, müşteri davranışlarını anlamak, pazarlama kampanyalarını optimize etmek veya ürün önerilerini kişiselleştirmek için farklı kaynaklardan (web sitesi, mobil uygulama, CRM, sosyal medya vb.) gelen verileri toplar, işler ve analiz eder. Bu süreç genellikle bir ETL (Extract, Transform, Load) veya ELT (Extract, Load, Transform) boru hattı gerektirir. Airflow, bu tür karmaşık boru hatlarını yönetmek için mükemmel bir araçtır.

Şimdi, e-ticaret verilerini işleyen basit bir ETL boru hattı senaryosunu Airflow ile nasıl otomatikleştireceğimizi inceleyelim.

Vaka Analizi: E-ticaret Müşteri Davranışları Analizi

Bir e-ticaret şirketi, günlük olarak farklı sistemlerden (web sitesi analizleri, sipariş veritabanı, pazarlama platformu) gelen verileri toplayıp, bunları birleştirerek müşteri segmentasyonu ve kampanya performans analizi yapmak istiyor. Hedef, bu verileri düzenli olarak bir veri ambarına (data warehouse) yüklemek ve BI (Business Intelligence) araçları aracılığıyla raporlanabilir hale getirmektir.

Boru Hattının Adımları:

1. Veri Çekme (Extract):
   • Günün web sitesi trafik verilerini bir harici API’den çekme.
   • Yeni sipariş verilerini şirket içi PostgreSQL veritabanından çekme.
2. Veri Dönüştürme (Transform):
   • Çekilen trafik ve sipariş verilerini birleştirme.
   • Müşteri ID’lerini anonimleştirme (hashleme).
   • Eksik değerleri doldurma veya aykırı değerleri temizleme.
   • Zaman damgalarını (timestamps) standart bir formata dönüştürme.
3. Veri Yükleme (Load):
   • İşlenmiş ve temizlenmiş veriyi şirketin veri ambarına (örneğin, Google BigQuery, Snowflake veya şirket içi bir PostgreSQL/Redshift) yükleme.
4. Raporlama Tetikleme (Optional):
   • Veri ambarına yükleme tamamlandıktan sonra, BI raporlama aracında (örneğin Tableau, Power BI) yeni raporların oluşturulmasını veya mevcut raporların yenilenmesini tetikleme.

Airflow ile Uygulama Örneği (Kısmi Kod):

Bu senaryo için e_ticaret_etl_dag.py adında bir DAG oluşturalım:

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.http.operators.http import SimpleHttpOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from airflow.models import Variable
from datetime import datetime, timedelta
import json
import pandas as pd # Veri işleme için

# --- Python Fonksiyonları ---

def web_traffic_api_cek(ti, **kwargs):
    # SimpleHttpOperator'dan gelen yanıtı çekiyoruz
    response_json = ti.xcom_pull(task_ids='web_traffic_cek', key='return_value')
    if not response_json:
        raise ValueError("Web trafik verisi API'den çekilemedi.")
    
    traffic_data = json.loads(response_json)
    print(f"Web trafik verisi çekildi: {len(traffic_data)} kayıt.")
    # Burada daha fazla işleme yapılabilir, şimdilik sadece yazdırıyoruz
    ti.xcom_push(key='web_traffic_data', value=traffic_data)

def siparis_verisi_cek(ti, **kwargs):
    pg_hook = PostgresHook(postgres_conn_id='postgres_default')
    sql_query = "SELECT order_id, customer_id, product_id, amount, order_date FROM orders WHERE order_date = CURRENT_DATE;"
    records = pg_hook.get_pandas_df(sql_query)
    print(f"PostgreSQL'den sipariş verisi çekildi: {len(records)} kayıt.")
    ti.xcom_push(key='order_data', value=records.to_json())

def verileri_donustur(ti, **kwargs):
    web_traffic_data = ti.xcom_pull(task_ids='web_traffic_api_isleme', key='web_traffic_data')
    order_data_json = ti.xcom_pull(task_ids='siparis_verisi_cek', key='order_data')
    
    if not web_traffic_data or not order_data_json:
        raise ValueError("Dönüştürme için gerekli veriler eksik.")

    order_df = pd.read_json(order_data_json)
    traffic_df = pd.DataFrame(web_traffic_data)

    # Örnek dönüşüm: Müşteri ID'lerini hashleme ve verileri birleştirme
    order_df['customer_id_hashed'] = order_df['customer_id'].apply(lambda x: hash(str(x)))
    # traffic_df ile order_df'yi birleştirmek için ortak bir anahtar olmalı
    # Burada basitçe örnek verilerle devam edelim
    
    merged_df = pd.merge(order_df, traffic_df, left_on='customer_id', right_on='user_id', how='left', suffixes=('_order', '_traffic'))
    merged_df = merged_df.fillna(0) # Eksik değerleri doldurma

    print(f"Veriler başarıyla dönüştürüldü. Birleştirilmiş kayıt sayısı: {len(merged_df)}")
    ti.xcom_push(key='transformed_data', value=merged_df.to_json())

def veriyi_veri_ambarina_yukle(ti, **kwargs):
    transformed_data_json = ti.xcom_pull(task_ids='verileri_donustur', key='transformed_data')
    if not transformed_data_json:
        raise ValueError("Yüklenecek dönüştürülmüş veri eksik.")

    final_df = pd.read_json(transformed_data_json)
    
    # Burada veri ambarına yükleme işlemleri yapılır.
    # Örneğin, BigQueryHook, SnowflakeHook veya başka bir veritabanı hook'u kullanılabilir.
    # Örnek olarak PostgreSQL'e yükleme yapalım:
    pg_hook = PostgresHook(postgres_conn_id='postgres_default')
    table_name = "customer_analytics_daily"
    final_df.to_sql(table_name, con=pg_hook.get_sqlalchemy_engine(), if_exists='append', index=False)
    
    print(f"{len(final_df)} kayıt '{table_name}' tablosuna başarıyla yüklendi.")

# --- DAG Tanımlaması ---

with DAG(
    dag_id='e_ticaret_etl_pipeline',
    start_date=datetime(2023, 1, 1),
    schedule_interval=timedelta(days=1), # Her gün çalışır
    catchup=False,
    tags=['etl', 'e-ticaret', 'veri-ambar'],
    default_args={
        'owner': 'airflow',
        'depends_on_past': False,
        'email_on_failure': False,
        'email_on_retry': False,
        'retries': 1,
        'retry_delay': timedelta(minutes=5),
    }
) as dag:
    # Görev 1: Web trafik verilerini API'den çekme
    web_traffic_cek = SimpleHttpOperator(
        task_id='web_traffic_cek',
        http_conn_id='http_default', # Airflow'da tanımlı HTTP bağlantısı
        endpoint='/api/v1/traffic', # Örnek API endpoint'i
        method='GET',
        headers={"Content-Type": "application/json"},
        response_filter=lambda response: response.text, # Yanıtı string olarak döndür
        log_response=True,
    )

    # Görev 1.1: Çekilen web trafik verisini işleme (XCom ile alma)
    web_traffic_api_isleme = PythonOperator(
        task_id='web_traffic_api_isleme',
        python_callable=web_traffic_api_cek,
    )

    # Görev 2: Sipariş verilerini PostgreSQL'den çekme
    siparis_verisi_cek = PythonOperator(
        task_id='siparis_verisi_cek',
        python_callable=siparis_verisi_cek,
    )

    # Görev 3: Çekilen verileri dönüştürme ve birleştirme
    verileri_donustur = PythonOperator(
        task_id='verileri_donustur',
        python_callable=verileri_donustur,
    )

    # Görev 4: Dönüştürülmüş veriyi veri ambarına yükleme
    veriyi_veri_ambarina_yukle = PythonOperator(
        task_id='veriyi_veri_ambarina_yukle',
        python_callable=veriyi_veri_ambarina_yukle,
    )

    # Görevler arası bağımlılıklar
    [web_traffic_cek, siparis_verisi_cek] >> verileri_donustur >> veriyi_veri_ambarina_yukle
  

Açıklamalar ve İpuçları:

1. Bağlantılar (Connections): Airflow, harici sistemlere (veritabanları, API’ler, bulut hizmetleri) bağlantı bilgilerini güvenli bir şekilde depolamanıza olanak tanır. http_default ve postgres_default gibi bağlantılar, Airflow UI’da “Admin -> Connections” altından tanımlanmalıdır. Bu, hassas bilgilerin (API anahtarları, veritabanı parolaları) kod içinde saklanmasını engeller.
2. XCom (Cross-Communication): Görevler arasında küçük miktarlarda veri (örneğin, dosya yolları, ID’ler, JSON yanıtları) aktarmak için kullanılır. xcom_push ile veri gönderilir, xcom_pull ile alınır. Yukarıdaki örnekte, API’den ve veritabanından çekilen veriler, işleme görevine XCom aracılığıyla aktarılır.
3. Operatörler (Operators):
   • SimpleHttpOperator: HTTP/HTTPS endpoint’lerine istek göndermek için kullanılır. API’den veri çekmek için idealdir.
   • PythonOperator: Özel Python fonksiyonlarınızı çalıştırmak için kullanılır. Veri dönüşümü ve veritabanı etkileşimleri için oldukça esnektir.
   • Diğer operatörler: Airflow, birçok farklı platform için hazır operatörler sunar (S3Hook, BigQueryOperator, SparkSubmitOperator vb.). Kendi ihtiyaçlarınıza göre bunları kullanabilir veya özel operatörler yazabilirsiniz.
4. Pandas Kullanımı: Python tabanlı veri işleme için Pandas kütüphanesi oldukça güçlüdür ve Airflow’daki PythonOperator içinde rahatlıkla kullanılabilir. Veri çerçevelerini (DataFrames) JSON’a dönüştürerek XCom üzerinden aktarmak yaygın bir yöntemdir.
5. Hata Yönetimi: default_args içinde retries ve retry_delay gibi parametreler tanımlayarak, görevlerin başarısız olması durumunda otomatik olarak yeniden denenmesini sağlayabilirsiniz. Ayrıca email_on_failure gibi seçeneklerle hata durumunda bildirim alabilirsiniz.

Bu vaka analizi, Airflow’un gerçek bir ETL senaryosunda nasıl kullanılabileceğine dair pratik bir bakış sunmaktadır. Karmaşık veri boru hatlarını yönetmek için Airflow’un modüler yapısı, esnek operatörleri ve güçlü izleme yetenekleri paha biçilmezdir.

Airflow’da İleri Düzey Konular ve En İyi Uygulamalar

Airflow’u temel seviyede kurup ilk DAG’lerinizi oluşturduktan sonra, üretim ortamlarında daha sağlam, verimli ve yönetilebilir iş akışları oluşturmak için bazı ileri düzey konulara ve en iyi uygulamalara hakim olmak önemlidir.

1. XCom (Cross-Communication) Kullanımı ve Sınırlamaları

XCom’lar, görevler arasında küçük miktarlarda veri paylaşmak için harika bir yoldur. Ancak, büyük veri kümelerini (örneğin, birkaç MB’tan fazla) XCom aracılığıyla aktarmaktan kaçınılmalısınız. Bunun yerine, büyük verileri paylaşılan bir depolama alanına (S3, GCS, HDFS) yazıp, XCom ile sadece verinin konumunu (dosya yolu, ID) aktarmak daha verimlidir.

2. Sensörler (Sensors)

Sensörler, belirli bir koşulun gerçekleşmesini bekleyen özel operatörlerdir. Örneğin, bir dosyanın S3’e yüklenmesini, bir veritabanı kaydının oluşmasını veya belirli bir API’nin yanıt vermesini bekleyebilirler. Bu, iş akışlarınızın harici olaylara tepki vermesini sağlar.

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

s3_dosya_bekle = S3KeySensor(
    task_id='s3_dosya_geldi_mi',
    bucket_name='veri-kovasi',
    bucket_key='input/yeni_veri.csv',
    aws_conn_id='aws_default',
    poke_interval=60, # Her 60 saniyede bir kontrol et
    timeout=60 * 60, # 1 saat sonra zaman aşımına uğra
)
  

3. Bağlantılar (Connections) ve Değişkenler (Variables)

• Bağlantılar: Veritabanı kimlik bilgileri, API anahtarları gibi hassas bilgileri Airflow UI’dan veya CLI aracılığıyla güvenli bir şekilde yönetin. Kodunuzda asla sabit kodlanmış kimlik bilgileri kullanmayın.
• Değişkenler: Ortam yapılandırmaları, eşikler veya sık kullanılan değerler gibi statik veya yarı statik verileri Airflow UI’dan veya CLI aracılığıyla depolayın. Bu, DAG’lerinizi daha dinamik hale getirir ve kodda değişiklik yapmadan yapılandırmaları güncellemenizi sağlar.

  
  from airflow.models import Variable
  
  # Airflow UI'da tanımlanmış bir değişkene erişim
  api_url = Variable.get("my_api_base_url", default_var="http://default-api.com")
            

4. Hata Yönetimi ve Bildirimler

• Yeniden Denemeler (Retries): Geçici hatalar için görevlere yeniden deneme sayıları (retries) ve gecikmeleri (retry_delay) ekleyin.
• E-posta/Slack Bildirimleri: default_args içinde email_on_failure=True veya Slack operatörleri gibi bildirim mekanizmalarını kullanarak görev başarısızlıklarında veya yeniden denemelerde ilgili ekipleri bilgilendirin.
• On-failure Callback’ler: Daha karmaşık hata işleme senaryoları için, bir görevin başarısız olması durumunda çalıştırılacak özel Python fonksiyonları tanımlayabilirsiniz (on_failure_callback).

5. Idempotency (Tekrarlanabilirlik)

Üretim ortamındaki veri boru hatlarının en önemli özelliklerinden biri tekrarlanabilirliktir. Bir görevin birden fazla kez çalıştırılması durumunda aynı sonucu üretmesi veya sistemin durumunu değiştirmemesi anlamına gelir. Örneğin, bir dosya yükleme görevi, dosya zaten varsa hata vermemeli veya aynı veriyi tekrar yüklememeli, bunun yerine var olanı güncellemelidir. Bu, hata durumunda görevleri güvenle yeniden çalıştırmanıza olanak tanır.

6. CI/CD Entegrasyonu

DAG’lerinizi sürüm kontrol sistemlerinde (Git) tutmak ve sürekli entegrasyon/sürekli dağıtım (CI/CD) süreçlerine dahil etmek, DAG geliştirme ve dağıtımını otomatikleştirmenin anahtarıdır. Yeni bir DAG veya DAG değişikliği Git’e gönderildiğinde, CI/CD boru hattı otomatik olarak DAG’i test edebilir ve Airflow’un DAG klasörüne dağıtabilir. Bu, hataları erken yakalar ve dağıtım sürecini hızlandırır.

7. Airflow Best Practices (En İyi Uygulamalar)

• Küçük ve Odaklanmış Görevler: Her görevin tek bir sorumluluğu olmalıdır (Single Responsibility Principle). Bu, hataları ayıklamayı ve yeniden denemeleri kolaylaştırır.
• Parametrik DAG’ler: Variable veya op_kwargs kullanarak DAG’lerinizi parametrik hale getirin. Bu, aynı DAG’i farklı girdilerle yeniden kullanmanızı sağlar.
• Test Edilebilirlik: DAG’lerinizi ve özel operatörlerinizi/hook’larınızı birim testleri (unit tests) ve entegrasyon testleri ile test edin.
• Günlük Kayıtları (Logging): Görevlerinizde yeterli ve anlamlı günlük kayıtları tutun. Airflow UI’dan bu günlükleri inceleyerek sorunları hızlıca teşhis edebilirsiniz.
• Resource Management: Özellikle KubernetesExecutor kullanırken, görevleriniz için CPU ve bellek sınırları belirleyerek kaynak tüketimini kontrol altında tutun ve sistem kararlılığını sağlayın.

Bu ileri düzey konular ve en iyi uygulamalar, Airflow ile daha güçlü, esnek ve güvenilir veri boru hatları oluşturmanıza yardımcı olacaktır. Airflow ekosistemi sürekli geliştiği için, topluluğu takip etmek ve yeni özellikleri öğrenmek de önemlidir.

Sonuç ve Gelecek Perspektifi

Bu makalede, Python’dan üretim hattına geçişte Apache Airflow’un neden bu kadar kritik bir rol oynadığını, temel bileşenlerini, yerel bir ortamda nasıl kurulacağını ve basit bir DAG’in nasıl oluşturulacağını adım adım inceledik. Ayrıca, gerçek dünya senaryolarından bir vaka analizi ile Airflow’un pratik kullanımını gösterdik ve ileri düzey konular ile en iyi uygulamalara değindik.

Airflow, veri odaklı şirketlerin karmaşık veri iş akışlarını otomatikleştirme, izleme ve yönetme biçimini devrim niteliğinde değiştirdi. Python’ın esnekliği ile birleşen güçlü orkestrasyon yetenekleri sayesinde, geliştiriciler ve veri mühendisleri, manuel hataları azaltırken veri boru hatlarının güvenilirliğini ve ölçeklenebilirliğini artırabilirler. İster küçük bir startup olun ister büyük bir kurumsal şirket, Airflow veri altyapınızın temel bir parçası haline gelerek operasyonel verimliliğinizi artırabilir ve değerli içgörülere daha hızlı ulaşmanızı sağlayabilir.

Gelecekte, Airflow’un bulut entegrasyonlarının daha da derinleştiğini, makine öğrenimi iş akışları (MLOps) için daha fazla özel operatör ve özellik sunduğunu ve genel olarak veri orkestrasyonu alanında standart olmaya devam ettiğini göreceğiz. Airflow topluluğu aktif ve büyümeye devam ediyor, bu da platformun sürekli gelişimi için sağlam bir zemin oluşturuyor. Python bilginizi kullanarak Airflow ile veri dünyasında harikalar yaratmaya başlamak için şimdi tam zamanı!

Sıkça Sorulan Sorular (SSS)

1. Apache Airflow’u kullanmak için Python bilmek şart mı?

Evet, Apache Airflow’un iş akışları (DAG’ler) Python ile tanımlandığı için Python programlama diline hakim olmak kesinlikle gereklidir. Temel Python bilgisi, Airflow’u etkili bir şekilde kullanmak için yeterlidir, ancak daha karmaşık görevler ve özel operatörler geliştirmek için ileri düzey Python bilgisi faydalı olacaktır.

2. Airflow, ETL/ELT süreçleri için tek çözüm müdür?

Hayır, Airflow ETL/ELT süreçleri için popüler ve güçlü bir araç olsa da tek çözüm değildir. Apache Nifi, Luigi, Prefect, Dagster gibi başka iş akışı yönetim araçları da mevcuttur. Airflow’un avantajları arasında geniş topluluk desteği, zengin operatör kütüphanesi ve Python tabanlı olması sayılabilir.

3. Airflow’u üretim ortamında kullanmak için nelere dikkat etmeliyim?

Üretim ortamında Airflow kullanırken ölçeklenebilirlik (CeleryExecutor veya KubernetesExecutor), yüksek erişilebilirlik, güvenlik (bağlantıların ve değişkenlerin şifrelenmesi), izleme ve uyarı sistemleri, CI/CD entegrasyonu ve düzenli yedeklemeler gibi konulara dikkat etmelisiniz. Ayrıca, görevlerinizin tekrarlanabilir (idempotent) olduğundan emin olun.

4. Airflow’daki ‘operatör’ ve ‘sensör’ arasındaki fark nedir?

Operatörler, bir görevin temel eylemini tanımlar; örneğin bir Bash komutu çalıştırmak (BashOperator), bir Python fonksiyonu çağırmak (PythonOperator) veya bir SQL sorgusu yürütmek (PostgresOperator). Sensörler ise, belirli bir koşulun gerçekleşmesini bekleyen özel operatörlerdir; örneğin bir dosyanın varlığını kontrol etmek (S3KeySensor) veya bir API’nin yanıt vermesini beklemek. Sensörler, koşul gerçekleşene kadar sürekli olarak “yoklama” (poke) yapar.

5. Airflow ile hangi bulut platformlarında çalışabilirim?

Apache Airflow, bulut platformlarından bağımsız olarak çalışabilir. AWS (Amazon Web Services), GCP (Google Cloud Platform) ve Azure gibi büyük bulut sağlayıcıları, Airflow için özel yönetilen hizmetler (örneğin Amazon MWAA, Google Cloud Composer) sunar. Bu hizmetler, Airflow kurulumu ve yönetimi yükünü azaltarak altyapı yerine DAG geliştirmeye odaklanmanızı sağlar. Kendi bulut sanal sunucularınızda veya Kubernetes kümelerinizde de Airflow’u manuel olarak kurup çalıştırabilirsiniz.

#ApacheAirflow #Python #VeriMühendisliği #ETL #Otomasyon #VeriBoruHattı