C2: Bir veri kalitesi ajanı yaz — DE Zoomcamp (Gençler)

Uykucu Gözlemci Problemi

Bir veri ekibinde klasik incident: “3 gündür marketing pipeline’ı sessizce boş geliyordu, dashboard’lar yanlış rapor yayınladı, müşteri iletişimi etkilendi.” Neden fark edilmedi? Çünkü kimse 7/24 tabloya bakmıyor.

Çözüm: veri kalitesi ajanı — schedule’lı çalışan, anomaly tespit eden, alert gönderen bir script. Bu capstone’ta production kalitesinde, gerçek bir uyarı kuralı ile çalışan bir ajan yazacaksınız. 80-150 satır Python, public repo, GitHub Actions’ta saatlik çalışır.

Adım 1: Anomali Kuralını Tanımla

Kural, “şu olursa haber ver” cümlesidir. Üç katman düşünün:

Katman	Örnek kural	Tip
Volume	”Dünkü satış, 7 günlük ortalamanın %50 altına düştüyse”	Ani düşüş
Schema	”Tablodaki sütun sayısı beklenenden farklıysa”	Yapısal bozulma
Freshness	”Son veri 6 saatten eski ise”	Pipeline durmuş
Distribution	”Ortalama fiyat, geçen haftaya göre 3 std sapma dışına çıktıysa”	İstatistiksel

Tek kural seçin, onu mükemmelleştirin. Çok kural = spagetti. Kural seçimi için README.md:

## Rule
- Metric: row count of `raw.sales` (yesterday)
- Baseline: 7-day rolling average
- Threshold: drop &gt; 50% from baseline OR z-score &lt; -2
- Action: send email + Slack webhook

## Why this rule
- Sudden 50% drop in sales almost never happens organically.
- Top causes: ingestion error, source API down, credential expired.
- Cost of false positive: 1 minute of investigation.
- Cost of false negative: hours of wrong reporting + customer trust.

Adım 2: Ajanı Yaz (production kalitesi)

dq_agent.py — type hints, logging, exit codes:

"""
Data Quality Agent — anomaly detection for raw.sales table.

Runs on schedule (cron / GitHub Actions / Kestra).
Fails (exit 1) on threshold breach, alerting via email + Slack.
"""
from __future__ import annotations

import logging
import os
import smtplib
import sys
from dataclasses import dataclass
from datetime import date, timedelta
from email.message import EmailMessage
from typing import Optional

import duckdb
import requests

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)
log = logging.getLogger("dq_agent")

# ---------- Config ----------
DB_PATH = os.environ.get("DUCKDB_PATH", "limonata.duckdb")
ALERT_FROM = os.environ["ALERT_FROM"]
ALERT_TO = os.environ["ALERT_TO"]
ALERT_APP_PASSWORD = os.environ["ALERT_APP_PASSWORD"]
SLACK_WEBHOOK = os.environ.get("SLACK_WEBHOOK_URL")  # optional

DROPPERCENT_THRESHOLD = 0.5
ZSCORE_THRESHOLD = -2.0
WINDOW_DAYS = 7
MIN_BASELINE_ROWS = 100  # avoid alerting on cold start


@dataclass
class CheckResult:
    name: str
    passed: bool
    detail: str


# ---------- Checks ----------
def check_volume_anomaly(con: duckdb.DuckDBPyConnection) -> CheckResult:
    """Compare yesterday's row count to 7-day rolling average."""
    today = date.today()
    y_rows = con.execute(
        "SELECT COUNT(*) FROM raw.sales WHERE DATE(order_time) = ?",
        [today - timedelta(days=1)],
    ).fetchone()[0]

    base_rows = con.execute(
        """
        SELECT AVG(daily_count) FROM (
            SELECT DATE(order_time) AS d, COUNT(*) AS daily_count
            FROM raw.sales
            WHERE DATE(order_time) BETWEEN ? AND ?
            GROUP BY 1
        )
        """,
        [today - timedelta(days=WINDOW_DAYS + 1), today - timedelta(days=2)],
    ).fetchone()[0] or 0

    if base_rows &lt; MIN_BASELINE_ROWS:
        return CheckResult("volume", True, f"baseline too small ({base_rows:.0f})")

    drop_pct = 1 - (y_rows / base_rows)
    passed = drop_pct &lt; DROPPERCENT_THRESHOLD
    return CheckResult(
        "volume",
        passed,
        f"yesterday={y_rows}, baseline={base_rows:.0f}, drop={drop_pct:.0%}",
    )


# ---------- Alerting ----------
def send_email(subject: str, body: str) -> None:
    msg = EmailMessage()
    msg["Subject"] = subject
    msg["From"] = ALERT_FROM
    msg["To"] = ALERT_TO
    msg.set_content(body)
    with smtplib.SMTP_SSL("smtp.gmail.com", 465) as s:
        s.login(ALERT_FROM, ALERT_APP_PASSWORD)
        s.send_message(msg)


def send_slack(text: str) -> None:
    if not SLACK_WEBHOOK:
        return
    requests.post(SLACK_WEBHOOK, json={"text": text}, timeout=5)


# ---------- Main ----------
def main() -> int:
    con = duckdb.connect(DB_PATH, read_only=True)
    results = [check_volume_anomaly(con)]
    con.close()

    failures = [r for r in results if not r.passed]
    for r in results:
        log.info(f"[{'PASS' if r.passed else 'FAIL'}] {r.name}: {r.detail}")

    if failures:
        body = "\n".join(f"- {r.name}: {r.detail}" for r in failures)
        send_email(
            subject=f"[DQ ALERT] {len(failures)} check(s) failed",
            body=f"Data quality breach detected:\n\n{body}\n\nInvestigate: ingestion log, source API status, last deploy.",
        )
        send_slack(f":rotating_light: DQ alert — {len(failures)} failure(s). See email.")
        log.error(f"{len(failures)} check(s) FAILED; alert sent.")
        return 1

    log.info("All checks PASS.")
    return 0


if __name__ == "__main__":
    sys.exit(main())

Secrets: Gmail App Password (2FA açıkken myaccount.google.com/apppasswords). Slack incoming webhook (api.slack.com/messaging/webhooks). Hiçbiri repo’ya yazılmaz; sadece GitHub Secrets.

Adım 3: Public Repo + GitHub Actions

Repo yapısı:

dq-agent/
├── dq_agent.py
├── tests/
│   └── test_check_volume_anomaly.py
├── .env.example
├── .github/workflows/dq.yml
├── Dockerfile
├── Makefile
└── README.md

.github/workflows/dq.yml — saatlik cron + manuel trigger:

name: dq-agent
on:
  schedule:
    - cron: "7 * * * *"   # her saat 7. dakika (off-peak)
  workflow_dispatch:

jobs:
  run:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.11" }
      - run: pip install -r requirements.txt
      - run: python dq_agent.py
        env:
          DUCKDB_PATH: ${{ secrets.DUCKDB_PATH }}
          ALERT_FROM: ${{ secrets.ALERT_FROM }}
          ALERT_TO: ${{ secrets.ALERT_TO }}
          ALERT_APP_PASSWORD: ${{ secrets.ALERT_APP_PASSWORD }}
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}

Adım 4: Test Yaz

Pytest ile pure function test edin (DB’ye bağlanmadan):

from dq_agent import CheckResult
from unittest.mock import MagicMock

def test_volume_anomaly_passes():
    con = MagicMock()
    con.execute.return_value.fetchone.side_effect = [(50,), (200,)]
    r = check_volume_anomaly(con)
    assert r.passed
    assert "yesterday=50" in r.detail

def test_volume_anomaly_fails_on_80pct_drop():
    con = MagicMock()
    con.execute.return_value.fetchone.side_effect = [(20,), (200,)]
    r = check_volume_anomaly(con)
    assert not r.passed
    assert "drop=90%" in r.detail

Adım 5: Yayınla + Monitor Et

gh repo create dq-agent --public --source=. --description="Veri kalitesi ajanı"
README’de triggered by bölümü: “GitHub Actions her saat başı çalıştırır.”
İlk run’da bilerek fail et (DB’yi sil, threshold’u aş). Alert gelmeli, ekran görüntüsünü docs/alert.png olarak ekle.
GitHub repo Insights → Community’den health yüzdesini %100’e tamamla (LICENSE, description, topics, code of conduct).

Production’a Geçiş Yol Haritası

Şu anki hali işi gören bir ajan. Production için eklenebilecekler:

Bugün (capstone)	Production
Tek bir volume check	5+ check (volume, freshness, schema, distribution)
SMTP mail + Slack	PagerDuty / Opsgenie (on-call rotation)
DuckDB local	Snowflake / BigQuery remote
Saatlik cron	Sürekli monitor + alert routing
Manual threshold	ML-based anomaly detection (Prophet, IsolationForest)
Repoda config	Config-as-code (YAML) + hot reload

Ne Öğrendik?

Anomali tespiti (windowed average, z-score) veri kalitesinin temel taşı.
Defense in depth: schema contract (gelen veri) + post-load check (tablodaki veri) + freshness (pipeline çalışıyor mu).
GitHub Actions + Secrets ile secrets-as-code güvenliği.
Pure function + mock test ile veritabanı bağımlılığı olmadan unit test.
Public repo + Loom demo ile portfolyo hikayesi.

Sıradaki

Capstone C3: Açık kaynak projeye PR gönder →