Tek platform, üç katman: ingest → staging → reports

Veri Platformu Nedir?

Şu ana kadar parça parça araçlar gördük: Kestra ile ingest, dlt ile veri alımı, dbt ile dönüşüm, BigQuery/Snowflake ile warehouse, Metabase ile raporlama. Her biri ayrı bir teknoloji, ayrı bir konfigürasyon, ayrı bir hata yüzeyi. Production’da bu yaklaşım operasyonel borç üretir: kim ne zaman çalıştı, hangi sürüm başarısız oldu, hangi adım retry edildi?

Veri platformu, tüm bu parçaları tek bir yönetim katmanı altında toplar. Bruin, bu yılın en pratik açık kaynak veri platformlarından biri. Modül 5’in amacı: önceki modüllerin hepsini tek bir mimaride birleştirmek.

Üç Katmanlı Mimari

Endüstri standardı, medallion architecture (bronze → silver → gold) varyasyonu olan üç katmandır:

┌─────────────────────────────────────────────────┐
│  1. INGEST  — ham veri (bronze)                  │
│     dlt, Python, API, Kafka → ham tablo          │
└─────────────────────────────────────────────────┘
                       ↓
┌─────────────────────────────────────────────────┐
│  2. STAGING — temizlenmiş veri (silver)          │
│     SQL dönüşümleri, tip düzeltme, dedup         │
└─────────────────────────────────────────────────┘
                       ↓
┌─────────────────────────────────────────────────┐
│  3. REPORTS — iş değeri olan veri (gold)         │
│     KPI tabloları, mart'lar, dashboard sorguları │
└─────────────────────────────────────────────────┘

Bu ayrım kasıtlıdır:

• Ingest idempotent olmalı: aynı veriyi iki kez çekersen aynı sonucu alırsın.
• Staging upstream’den gelen şemaya güvenmemeli: kolon tipleri, NULL yönetimi, encode.
• Reports iş mantığı taşır: iş birimi bu katmandan okur, upstream’i bilmez.

Bruin Kurulumu

# Mac / Linux
brew install bruin-data/tap/bruin

# Cross-platform
curl -LsSf https://getbruin.com/install.sh | sh

# veya
pipx install bruin

bruin --version    # 0.10+ beklenir

VS Code extension: code --install-extension bruin.bruin veya marketplace’ten “Bruin” ara. Kurunca .bruin.yml olan bir klasör açtığında MCP (Model Context Protocol) sunucusu otomatik başlar. Bu, Claude Code’un Bruin pipeline’ını okumasını sağlar: hangi asset var, hangi connection kullanılıyor, hangi SQL bekleniyor — hepsi editör context’ine girer.

Chess.com API Starter Projesi

Bu dersin canlı örneği Chess.com public API’si. Ücretsiz, kayıt yok, JSON döner. Kendi capstone’ın için de uyarlayabilirsin (kullanıcı adı değiştir, başka endpoint ekle).

Proje yapısı:

chess-pipeline/
├── .bruin.yml
├── connections/
│   └── warehouse.postgres.yml
├── pipeline/
│   ├── assets/
│   │   ├── chess_games.py        # Ingest (Python asset)
│   │   ├── stg_games.sql         # Staging (SQL asset)
│   │   └── reports/
│   │       └── player_stats.sql  # Reports (SQL asset)
│   └── run.py
└── README.md

.bruin.yml

default_pipeline: chess-pipeline

pipelines:
  chess-pipeline:
    start_date: 2025-01-01
    schedule: "0 6 * * *"   # Her gün 06:00 UTC

assets/chess_games.py (ingest)

"""@bruin
name: chess.games.ingest
type: python
image: python:3.11
connection: warehouse

@bruin"""
import requests

def fetch(month: str, username: str = "magnus") -> list:
    url = f"https://api.chess.com/pub/player/{username}/games/{month}"
    headers = {"User-Agent": "de-zoomcamp-teens/1.0"}
    data = requests.get(url, headers=headers, timeout=30).json()
    return data.get("games", [])

Bruin asset tanımı dosyanın başındaki docstring içinde, declarative: ne, hangi tip, hangi image, hangi connection.

assets/stg_games.sql (staging)

/* @bruin
name: chess.games.staging
type: duckdb.sql
materialization: table
depends:
  - chess.games.ingest
@bruin */

SELECT
  game_id,
  player_username,
  opponent_username,
  result,
  CAST(end_time AS TIMESTAMP) AS ended_at,
  time_control,
  ECOopening   -- "C20" gibi kodlar
FROM chess.games.ingest
WHERE end_time IS NOT NULL

depends: satırı lineage’ı dosyada deklare eder — başka bir yere yazmana gerek yok.

assets/reports/player_stats.sql (reports)

/* @bruin
name: chess.player.stats
type: duckdb.sql
materialization: table
depends:
  - chess.games.staging
@bruin */

SELECT
  player_username,
  COUNT(*)                              AS games,
  SUM(CASE WHEN result = 'win'  THEN 1 ELSE 0 END) AS wins,
  SUM(CASE WHEN result = 'loss' THEN 1 ELSE 0 END) AS losses,
  SUM(CASE WHEN result = 'draw' THEN 1 ELSE 0 END) AS draws,
  ROUND(100.0 * SUM(CASE WHEN result = 'win' THEN 1 ELSE 0 END) / COUNT(*), 1) AS win_rate_pct
FROM chess.games.staging
GROUP BY player_username

Çalıştırma

bruin run pipeline/                  # Tüm pipeline
bruin run --start-date 2025-01-01 --end-date 2025-01-31 pipeline/   # Tarih aralığı
bruin run --only chess.player.stats pipeline/    # Sadece bu asset

Bruin paralel çalıştırmayı, incremental’i, dependency çözümünü, retry’i dahili yapar. Bu yüzden “platform” diyoruz — pipeline değil, pipeline yönetimi.

Hands-on

1. bruin --version çalışıyor mu doğrula.
2. VS Code extension’ı kur, bir klasörde .bruin.yml aç, MCP’nin aktif olduğunu gör.
3. Yukarıdaki chess-pipeline/ iskeletini oluştur, kendi kullanıcı adınla test et.
4. bruin run pipeline/ çalıştır, hata olursa oku, düzelt.
5. Bir SQL’i kasıtlı boz (yanlış kolon adı), lineage hatasının hangi asset’te patladığını gözlemle.

Sıradaki

Pipeline çalışıyor. Şimdi verinin izini lineage ile takip edelim →