ELT ve dbt mantığı: katmanlı SQL dönüşümleri ve testler

ETL'den ELT'ye

veri-boru-hatti dersinde veriyi Python'da çıkarıp dönüştürdün, sonra veritabanına yükledin (ETL: extract, transform, load). Bulut ambarları (BigQuery, Snowflake, Redshift) ve DuckDB gibi hızlı analitik motorlar çıkınca sıra değişti: ham veri önce olduğu gibi ambara yüklenir, dönüşüm ambarın içinde SQL ile yapılır. Buna ELT denir. Avantajı: ham veri her zaman elindedir (yeniden hesaplanabilir), dönüşüm ambarın paralel gücüyle çalışır ve iş mantığı analistlerin de okuyabildiği SQL'dedir.

ELT'de yüzlerce SQL dosyası birikir. Hangisi hangisinden önce çalışmalı? Hangi tablo hangi kaynaktan geliyor? Bir sütunu değiştirince ne bozulur? dbt (data build tool) bu soruları çözen açık kaynak araçtır: SQL dosyalarını model olarak düzenler, Jinja şablonlarıyla modüler hale getirir, aralarındaki bağımlılıklardan bir grafik çıkarır, doğru sırada çalıştırır, test eder ve belgeler. Bu derste dbt'nin mantığını öğrenip DuckDB + jinja2 ile kendi "mini-dbt"ni yazacak, sonunda kanat_havayollari için katmanlı bir proje kuracaksın.

dbt proje yapısı ve katmanlar

kanat/
├── dbt_project.yml          # proje adı, varsayılan materializasyonlar
├── models/
│   ├── staging/             # kaynakla 1-1: yeniden adlandır, tür dönüştür, temizle
│   │   ├── _sources.yml     # ham tablolar (source) tanımı
│   │   ├── stg_ucuslar.sql
│   │   └── stg_gecikmeler.sql
│   ├── intermediate/        # iş mantığı ara adımları (birleştir, taneciğe indir)
│   │   └── int_ucus_gecikme.sql
│   └── marts/               # son kullanıcıya hazır olgu ve boyutlar
│       ├── mart_rota_performans.sql
│       └── _marts.yml       # testler ve belgeler
├── seeds/                   # küçük sabit CSV'ler (kod tabloları)
├── snapshots/               # SCD2 anlık görüntüleri
└── tests/                   # özel (singular) SQL testleri

• sources: yükleme aracının ambara bıraktığı ham tablolar. Modeller ham tabloya adıyla değil {{ source('kanat', 'hv_flights') }} ile erişir.
• staging (stg_): her kaynak tablo için bir model. Sütunları yeniden adlandırır, türleri dönüştürür, birimleri düzeltir. JOIN yok, iş mantığı yok.
• intermediate (int_): staging modellerini birleştirir, taneciği değiştirir (ör. gecikme nedenlerini uçuş düzeyine toplar).
• marts (mart_, fct_, dim_): panolara ve analistlere açılan son tablolar; veri-modelleme dersindeki yıldız şema burada kurulur.

-- models/staging/stg_ucuslar.sql
select
    flight_id,
    flight_no                                as ucus_no,
    route_id,
    status                                   as durum,
    cast(sched_dep as timestamp)             as plan_kalkis,
    date_diff('minute', cast(sched_dep as timestamp), cast(actual_dep as timestamp)) as kalkis_gecikme_dk
from {{ source('kanat', 'hv_flights') }}

-- models/marts/mart_rota_performans.sql
{{ config(materialized='table') }}
select r.rota, count(*) as ucus,
       avg(case when u.kalkis_gecikme_dk <= 15 then 1 else 0 end) as zamaninda_oran
from {{ ref('stg_ucuslar') }} u
join {{ ref('stg_rotalar') }} r using (route_id)
where u.durum <> 'cancelled'
group by 1

ref() ve bağımlılık grafiği

dbt'nin kalbi ref() fonksiyonudur. Model içinde {{ ref('stg_ucuslar') }} yazdığında iki şey olur: (1) şablon derlenirken bu ifade gerçek tablo adına (analytics.stg_ucuslar) dönüşür; (2) dbt "bu model stg_ucuslar'a bağımlı" diye not alır. Tüm modellerin notlarından yönlü, döngüsüz bir grafik (DAG: directed acyclic graph) çıkar; dbt modelleri bu grafiğin topolojik sırasıyla çalıştırır. Tablo adını elle yazarsan bağımlılık kaybolur ve sıra bozulur. Bunu jinja2 ile kendin kurabilirsin:

from jinja2 import Template

def derle(sablon, sema="analytics"):
    bagimliliklar = []
    def ref(ad):
        bagimliliklar.append(ad)      # derlerken bağımlılığı kaydet
        return f"{sema}.{ad}"
    sql = Template(sablon).render(ref=ref)
    return sql, bagimliliklar

sablon = "select * from {{ ref('stg_ucuslar') }} u join {{ ref('stg_rotalar') }} r using (route_id)"
sql, bag = derle(sablon)
print(sql)
print(bag)

Python'un standart kütüphanesindeki graphlib.TopologicalSorter bağımlılık grafiğini sıralar ve döngü varsa CycleError verir:

from graphlib import TopologicalSorter

graf = {                              # model: [bağımlı olduğu modeller]
    "stg_ucuslar": [],
    "stg_rotalar": [],
    "stg_gecikmeler": [],
    "int_ucus_gecikme": ["stg_gecikmeler"],
    "mart_rota_performans": ["stg_ucuslar", "stg_rotalar", "int_ucus_gecikme"],
}
print(list(TopologicalSorter(graf).static_order()))

Jinja ile modüler SQL: makrolar

Jinja; değişken ({{ }}), kontrol yapısı ({% for %}, {% if %}) ve makro ({% macro %}) sağlar. Tekrar eden SQL'i (ör. kuruşu liraya çevirmek, para birimi sütunları) bir makroya alırsın; dbt paketleri (dbt_utils) yüzlerce hazır makro sunar.

from jinja2 import Template

sablon = """
{%- macro dk_to_saat(sutun) -%} round({{ sutun }} / 60.0, 2) {%- endmacro -%}
select
{%- for s in sutunlar %}
    {{ dk_to_saat(s) }} as {{ s }}_saat{{ "," if not loop.last }}
{%- endfor %}
from ucuslar
"""
print(Template(sablon).render(sutunlar=["blok_dk", "gecikme_dk"]))

Materializasyon: view, table, incremental

-- models/marts/fct_rezervasyon.sql
{{ config(materialized='incremental', unique_key='booking_id') }}
select * from {{ ref('stg_rezervasyonlar') }}
{% if is_incremental() %}
  -- yalnız hedefte olan en yeni kayıttan sonrakiler (watermark)
  where rezervasyon_zamani > (select max(rezervasyon_zamani) from {{ this }})
{% endif %}

Artımlı modelin üç riski vardır: geç gelen kayıtlar (watermark'tan eski zaman damgasıyla gelir, kaçırılır), güncellenen kayıtlar (unique_key ile upsert gerekir) ve şema değişikliği. Çözüm: watermark'a küçük bir geri bakma payı (lookback) eklemek, unique_key vermek ve ara sıra dbt run --full-refresh çalıştırmak.

Testler

dbt'de test, hatalı satırları döndüren bir SQL sorgusudur: 0 satır dönerse test geçer. Dört yerleşik genel (generic) test YAML'da tanımlanır:

# models/marts/_marts.yml
models:
  - name: stg_ucuslar
    description: "Her satır bir planlı uçuş."
    columns:
      - name: flight_id
        tests: [unique, not_null]
      - name: durum
        tests:
          - accepted_values: {values: ['landed', 'cancelled', 'diverted']}
      - name: route_id
        tests:
          - relationships: {to: ref('stg_rotalar'), field: route_id}

import duckdb
import veri

con = duckdb.connect()
con.execute(f"CREATE VIEW ucus AS SELECT * FROM read_csv('{veri.yol('kanat_havayollari', 'hv_flights')}')")
TESTLER = {
    "unique": "select {s} from {t} group by {s} having count(*) > 1",
    "not_null": "select * from {t} where {s} is null",
}
for tur, sablon in TESTLER.items():
    for sutun in ["flight_id", "actual_dep"]:
        hatali = con.execute(f"select count(*) from ({sablon.format(t='ucus', s=sutun)})").fetchone()[0]
        print(f"{tur:9} {sutun:11} -> {'GEÇTİ' if hatali == 0 else f'KALDI ({hatali})'}")

actual_dep üzerindeki not_null testi 62 iptal uçuş yüzünden kalır. Bu bir veri hatası değil, kuralın eksikliğidir: test where: "status <> 'cancelled'" koşuluyla yazılmalıdır. Testler, iş kuralını açıkça yazmaya zorlar. Özel (singular) testler tests/ klasöründe tek bir SQL dosyasıdır, ör. "iniş zamanı kalkıştan önce olamaz".

Belgeleme ve köken (lineage)

dbt docs generate YAML'daki açıklamalardan ve bağımlılık grafiğinden bir web sitesi üretir: her modelin sütunları, testleri ve köken grafiği. Köken grafiği iki soruyu yanıtlar: "Bu sayı nereden geliyor?" (yukarı akış, upstream) ve "Bu tabloyu değiştirirsem ne bozulur?" (aşağı akış, downstream: etki analizi). Seçici sözdizimi de bu grafiğe dayanır: dbt run --select +mart_rota_performans modeli ve tüm atalarını, stg_ucuslar+ modeli ve tüm torunlarını çalıştırır.

Snapshot ve seed

Snapshot, kaynak tablodaki değişiklikleri SCD tip 2 olarak saklar (veri-modelleme dersi). dbt her çalışmada izlenen sütunları karşılaştırır, değişen satırın eski sürümüne dbt_valid_to yazar ve yeni sürüm ekler. Seed ise depoda tutulan küçük CSV dosyalarıdır (gecikme neden kodları, ülke-bölge eşlemesi); dbt seed onları tabloya yükler ve modeller ref() ile kullanır.

-- snapshots/yolcu_snapshot.sql
{% snapshot yolcu_snapshot %}
{{ config(target_schema='snapshots', unique_key='passenger_id',
          strategy='check', check_cols=['tier', 'city']) }}
select * from {{ source('kanat', 'hv_passengers') }}
{% endsnapshot %}