ETL nedir? Veriyi taşımanın sistematik yolu

2026 · data engineering · 18 dakika okuma

Veri nerede üretilir? Genellikle üretim veritabanlarında, API'lerde, dosyalarda, log sistemlerinde. Analiz nerede yapılır? Veri ambarında, BI araçlarında, Jupyter notebook'ta. Bu iki nokta arasındaki köprü ETL.

ETL nedir?

ETL üç adımın kısaltması: Extract (Çıkar), Transform (Dönüştür), Load (Yükle). Ham veriyi kaynak sistemden alıp, analiz için hazır hale getirip, hedef sisteme yüklemek.

Neden ETL gerekli?

Çoğu şirkette veri farklı sistemlerde dağınık halde bulunur. Satış verisi CRM'de, ürün verisi ERP'de, web trafiği Google Analytics'te, müşteri yorumları ayrı bir platformda. Bunları birleştirmeden anlamlı analiz yapamazsın.

Üstelik üretim veritabanları analiz için optimize edilmemiştir. Sorgu çalıştırınca canlı sistemi yavaşlatırsın. ETL ile veriyi ayrı bir analiz ortamına taşırsın.

1. Extract — Veriyi çekme

Kaynak sistemden veriyi almak. En yaygın senaryolar:

import pandas as pd
import sqlite3
import requests

# ── Veritabanından çekme ──────────────────────────────
conn = sqlite3.connect('uretim.db')
df_siparisler = pd.read_sql("""
    SELECT s.id, s.tarih, s.musteri_id, s.toplam
    FROM siparisler s
    WHERE s.tarih >= '2024-01-01'
""", conn)
conn.close()

# ── CSV / Excel'den çekme ─────────────────────────────
df_urunler = pd.read_csv('urunler.csv')
df_butce   = pd.read_excel('butce_2024.xlsx', sheet_name='Q1')

# ── API'den çekme ─────────────────────────────────────
response = requests.get(
    'https://api.orneksite.com/musteriler',
    headers={'Authorization': 'Bearer TOKEN'},
    params={'limit': 1000, 'offset': 0}
)
df_musteriler = pd.DataFrame(response.json()['data'])

print(f"Siparişler: {df_siparisler.shape}")
print(f"Ürünler:    {df_urunler.shape}")
print(f"Müşteriler: {df_musteriler.shape}")

2. Transform — Dönüştürme

ETL'nin kalbi burada. Ham veriyi analiz için hazır hale getirmek: temizleme, birleştirme, hesaplama, standardize etme.

import pandas as pd
import numpy as np

# ── Temel temizleme ───────────────────────────────────
def temizle(df):
    df = df.copy()

    # Sütun adlarını standardize et
    df.columns = (df.columns
        .str.strip()
        .str.lower()
        .str.replace(' ', '_'))

    # Tekrar eden satırları sil
    df = df.drop_duplicates()

    # Tarih sütunlarını dönüştür
    for col in df.columns:
        if 'tarih' in col or 'date' in col:
            df[col] = pd.to_datetime(df[col], errors='coerce')

    return df

# ── İş kuralları uygula ───────────────────────────────
def is_kurallari_uygula(df_siparisler, df_urunler):
    # Birleştir
    df = df_siparisler.merge(
        df_urunler[['id', 'kategori', 'maliyet']],
        left_on='urun_id',
        right_on='id',
        how='left'
    )

    # Hesaplamalar
    df['kar'] = df['satis_fiyati'] - df['maliyet']
    df['kar_marji'] = (df['kar'] / df['satis_fiyati'] * 100).round(2)
    df['ay'] = df['siparis_tarihi'].dt.to_period('M')
    df['yil'] = df['siparis_tarihi'].dt.year

    # Kategorik dönüşüm
    df['segment'] = pd.cut(
        df['toplam'],
        bins=[0, 500, 2000, 10000, float('inf')],
        labels=['Küçük', 'Orta', 'Büyük', 'Kurumsal']
    )

    return df

# ── Aggregation (özetleme) ────────────────────────────
def aylik_ozet_olustur(df):
    return df.groupby(['ay', 'kategori']).agg(
        siparis_sayisi=('id', 'count'),
        toplam_ciro=('toplam', 'sum'),
        ort_siparis=('toplam', 'mean'),
        toplam_kar=('kar', 'sum'),
        ort_kar_marji=('kar_marji', 'mean'),
    ).round(2).reset_index()

3. Load — Hedefe yükleme

Dönüştürülmüş veriyi analiz ortamına yükle. Hedef bir SQL veritabanı, Parquet dosyası, Google BigQuery veya BI aracı olabilir.

import pandas as pd
import sqlite3

# ── SQLite'a yükle (geliştirme/test) ─────────────────
def sqlite_yukle(df, tablo_adi, db_yolu='analiz.db'):
    conn = sqlite3.connect(db_yolu)
    df.to_sql(
        tablo_adi,
        conn,
        if_exists='replace',   # 'replace', 'append', 'fail'
        index=False,
        chunksize=1000         # büyük veri için toplu yükleme
    )
    conn.close()
    print(f"✓ {len(df):,} satır → {tablo_adi}")

# ── Parquet'a yükle (hızlı, sıkıştırılmış) ───────────
def parquet_yukle(df, dosya_yolu):
    df.to_parquet(
        dosya_yolu,
        engine='pyarrow',
        compression='snappy',
        index=False
    )
    print(f"✓ Kaydedildi: {dosya_yolu}")

# ── CSV olarak arşivle ────────────────────────────────
def arsivle(df, klasor='arsiv'):
    import os
    from datetime import datetime
    os.makedirs(klasor, exist_ok=True)
    tarih = datetime.now().strftime('%Y%m%d_%H%M')
    yol = f"{klasor}/veri_{tarih}.csv"
    df.to_csv(yol, index=False)
    print(f"✓ Arşivlendi: {yol}")

Tam ETL pipeline

Gerçek dünyada tüm adımları birleştiren, hata yönetimine sahip, loglama yapan bir pipeline şöyle görünür:

import pandas as pd
import sqlite3
import logging
from datetime import datetime

# Loglama ayarla
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
log = logging.getLogger(__name__)

def etl_calistir():
    baslangic = datetime.now()
    log.info("ETL pipeline başladı")

    try:
        # ── EXTRACT ──────────────────────────────────
        log.info("Veri çekiliyor...")
        conn = sqlite3.connect('uretim.db')
        df = pd.read_sql(
            "SELECT * FROM siparisler WHERE tarih >= date('now', '-30 days')",
            conn
        )
        conn.close()
        log.info(f"  ✓ {len(df):,} satır çekildi")

        # ── TRANSFORM ────────────────────────────────
        log.info("Dönüştürme başlıyor...")

        # Temizle
        df = df.drop_duplicates()
        df['tarih'] = pd.to_datetime(df['tarih'])
        df['toplam'] = pd.to_numeric(df['toplam'], errors='coerce')
        df = df.dropna(subset=['toplam', 'tarih'])

        # Hesapla
        df['ay'] = df['tarih'].dt.to_period('M').astype(str)
        df['hafta'] = df['tarih'].dt.isocalendar().week

        # Özetle
        ozet = df.groupby('ay').agg(
            siparis_sayisi=('id', 'count'),
            toplam_ciro=('toplam', 'sum'),
        ).reset_index()

        log.info(f"  ✓ {len(df):,} satır dönüştürüldü")

        # ── LOAD ──────────────────────────────────────
        log.info("Yükleniyor...")
        analiz_conn = sqlite3.connect('analiz.db')
        df.to_sql('siparisler_temiz', analiz_conn,
                  if_exists='replace', index=False)
        ozet.to_sql('aylik_ozet', analiz_conn,
                    if_exists='replace', index=False)
        analiz_conn.close()
        log.info("  ✓ Yükleme tamamlandı")

        # Özet rapor
        sure = (datetime.now() - baslangic).seconds
        log.info(f"ETL tamamlandı — {sure}s | {len(df):,} satır")
        return True

    except Exception as e:
        log.error(f"ETL HATASI: {e}")
        raise

if __name__ == '__main__':
    etl_calistir()

ETL vs ELT

Modern veri ambarları (BigQuery, Snowflake, Redshift) çok güçlü. Bu yüzden son yıllarda ELT (Extract, Load, Transform) yaygınlaştı — önce ham veriyi yükle, dönüşümü ambar içinde yap.

Popüler ETL araçları

• Apache Airflow — Python tabanlı workflow orkestrasyon. Büyük şirketlerin tercihi.
• dbt — Transform katmanı için. SQL ile modelleme, versiyon kontrolü.
• Prefect — Airflow'dan daha modern, Python odaklı.
• Luigi — Spotify'ın açık kaynak pipeline aracı. Daha basit kurulum.
• Pandas — Küçük-orta ölçekli ETL için yeterli. Öğrenmesi en kolay.

Altın kurallar

• Ham veriyi koru. Dönüşüm öncesi her zaman bir kopyası olsun. Bir hata yapınca geri dönebilmek için.
• Her adımı logla. Kaç satır çekildi, kaç satır düştü, ne kadar sürdü. Sorun çıkınca neyin nerede bozulduğunu bilmek için.
• Küçük başla. 1 milyon satırda test etme. Önce 1000 satırla doğrula.
• Idempotent yap. Pipeline'ı iki kez çalıştırsan aynı sonucu vermeli. Veriyi iki kez yüklememeli.
• Hata yönetimini unutma. Kaynak sistem düşebilir, API cevap vermeyebilir. Her adım için try/except yaz.

Sıradaki yazıda dbt ile analytics engineering: SQL ile transform katmanı nasıl yazılır, veri modelleri nasıl oluşturulur.