Orkestratsiya: quvurni bosqichlarga bo'lish

1. ma'lumotni o'qish        (2 daqiqa)
2. tozalash                 (5 daqiqa)
3. xususiyat hisoblash      (30 daqiqa)
4. model o'qitish           (15 daqiqa)
5. baholash va saqlash      (1 daqiqa)

4-bosqich: ValueError

import pandas as pd
show(pd.DataFrame([
    {"muammo": "Oxirgi bosqichda yiqildi", "bitta skriptda": "Hammasi boshidan qayta ishlaydi",
     "orkestratsiyada": "Faqat yiqilgan bosqichdan davom etadi"},
    {"muammo": "Faqat modelni almashtirmoqchisiz", "bitta skriptda": "Ma'lumot ham qayta tayyorlanadi",
     "orkestratsiyada": "Tayyorlash keshdan olinadi"},
    {"muammo": "Ikki bosqich bir-biriga bog'liq emas", "bitta skriptda": "Ketma-ket bajariladi",
     "orkestratsiyada": "Parallel bajarilishi mumkin"},
    {"muammo": "Kecha nima bo'lgan edi?", "bitta skriptda": "Jurnalda qidirasiz",
     "orkestratsiyada": "Har bosqichning holati saqlanadi"},
]))

diagramma(
    {
        "1. oqi\n(CSV -> parquet)": (0, 2.0, 2.5, 1.0, "manba"),
        "2. tekshir\n(sxema, sifat)": (3.2, 2.0, 2.5, 1.0, "qayta"),
        "3. tayyorla\n(xususiyatlar)": (6.4, 2.0, 2.5, 1.0, "qayta"),
        "4. oqit\n(model)": (9.6, 0.8, 2.5, 1.0, "qayta"),
        "5. baho\n(metrikalar)": (9.6, 3.2, 2.5, 1.0, "kuzatuv"),
    },
    [
        ("1. oqi\n(CSV -> parquet)", "2. tekshir\n(sxema, sifat)"),
        ("2. tekshir\n(sxema, sifat)", "3. tayyorla\n(xususiyatlar)"),
        ("3. tayyorla\n(xususiyatlar)", "4. oqit\n(model)"),
        ("3. tayyorla\n(xususiyatlar)", "5. baho\n(metrikalar)"),
    ],
    nom="quvur_dag.png", kenglik=13,
    izoh="Har bosqich o'z chiqishini yozadi. Keyingi bosqich o'sha faylni kirish sifatida oladi.",
)

"""Oddiy DAG ishga tushirgich: bog'liqlik, keshlash, xato boshqaruvi.

Airflow shuni katta miqyosda qiladi. Bu yerda g'oyani ko'rish uchun eng kichik variant.
"""
import json
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable


@dataclass
class Bosqich:
    nom: str
    ish: Callable[[], Any]
    kirish: tuple[str, ...] = ()              # kerak bo'ladigan fayllar
    chiqish: tuple[str, ...] = ()             # yasaydigan fayllar
    bogliq: tuple[str, ...] = ()              # oldin bajarilishi kerak bo'lgan bosqichlar


@dataclass
class Natija:
    holat: str                                # "bajarildi" | "keshdan" | "yiqildi" | "o'tkazildi"
    vaqt_s: float = 0.0
    xato: str = ""


def _yangimi(kirish: tuple[str, ...], chiqish: tuple[str, ...]) -> bool:
    """Chiqish fayllari kirishdan yangimi? Yangi bo'lsa qayta bajarish shart emas."""
    if not chiqish:
        return False
    ch = [Path(c) for c in chiqish]
    if not all(c.exists() for c in ch):
        return False
    ki = [Path(k) for k in kirish if Path(k).exists()]
    if not ki:
        return True
    return min(c.stat().st_mtime for c in ch) >= max(k.stat().st_mtime for k in ki)


def tartibla(bosqichlar: list[Bosqich]) -> list[Bosqich]:
    """Bog'liqlik bo'yicha tartiblaydi (topologik). Sikl bo'lsa xato beradi."""
    qolgan = {b.nom: b for b in bosqichlar}
    tayyor: list[Bosqich] = []
    korilgan: set[str] = set()
    while qolgan:
        ozod = [b for b in qolgan.values() if all(d in korilgan for d in b.bogliq)]
        if not ozod:
            raise ValueError(f"sikl yoki yetishmayotgan bog'liqlik: {sorted(qolgan)}")
        for b in sorted(ozod, key=lambda x: x.nom):
            tayyor.append(b)
            korilgan.add(b.nom)
            del qolgan[b.nom]
    return tayyor


def ishga_tushir(bosqichlar: list[Bosqich], majburiy: bool = False,
                 holat_yoli: str | Path = "quvur_holati.json") -> dict[str, Natija]:
    """DAG ni bajaradi. Bosqich yiqilsa, unga bog'liqlar O'TKAZIB YUBORILADI."""
    natijalar: dict[str, Natija] = {}
    for b in tartibla(bosqichlar):
        yiqilgan = [d for d in b.bogliq
                    if natijalar.get(d) and natijalar[d].holat in {"yiqildi", "o'tkazildi"}]
        if yiqilgan:
            natijalar[b.nom] = Natija("o'tkazildi", xato=f"bog'liq yiqildi: {yiqilgan[0]}")
            continue
        if not majburiy and _yangimi(b.kirish, b.chiqish):
            natijalar[b.nom] = Natija("keshdan")
            continue
        t0 = time.perf_counter()
        try:
            b.ish()
            natijalar[b.nom] = Natija("bajarildi", round(time.perf_counter() - t0, 3))
        except Exception as e:
            natijalar[b.nom] = Natija("yiqildi", round(time.perf_counter() - t0, 3),
                                      f"{type(e).__name__}: {e}")
    Path(holat_yoli).write_text(
        json.dumps({k: v.__dict__ for k, v in natijalar.items()}, ensure_ascii=False, indent=2),
        encoding="utf-8")
    return natijalar


def chop_et(natijalar: dict[str, Natija]) -> None:
    BELGI = {"bajarildi": "+", "keshdan": "=", "yiqildi": "X", "o'tkazildi": "-"}
    for nom, n in natijalar.items():
        vaqt = f"{n.vaqt_s:6.2f}s" if n.vaqt_s else "      -"
        print(f"  [{BELGI[n.holat]}] {nom:12} {n.holat:12} {vaqt}  {n.xato}")

"""hisobot quvuri: oqi -> tekshir -> tayyorla -> oqit -> baho."""
import json
import sys
from pathlib import Path

sys.path.insert(0, "src")

import joblib
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import f1_score, precision_score, recall_score

from quvur.dag import Bosqich, chop_et, ishga_tushir
from quvur.sxema import IJARA_SXEMASI, tekshir as sxema_tekshir
from quvur.xususiyat import XUSUSIYATLAR, tayyorla

XOM = "../datasets/ijara.csv"
ORALIQ = Path("oraliq")
ORALIQ.mkdir(exist_ok=True)

P_XOM = ORALIQ / "xom.parquet"
P_TOZA = ORALIQ / "toza.parquet"
P_XUS = ORALIQ / "xususiyat.parquet"
P_MODEL = ORALIQ / "model.joblib"
P_BAHO = ORALIQ / "baho.json"


def qadam_oqi():
    pd.read_csv(XOM).to_parquet(P_XOM, index=False)


def qadam_tekshir():
    toza, karantin, _ = sxema_tekshir(pd.read_parquet(P_XOM), IJARA_SXEMASI)
    toza.to_parquet(P_TOZA, index=False)
    if len(karantin):
        karantin.to_parquet(ORALIQ / "karantin.parquet", index=False)


def qadam_tayyorla():
    d = pd.read_parquet(P_TOZA)
    x = tayyorla(d)
    x["broker"] = d["broker"].values
    x.to_parquet(P_XUS, index=False)


def qadam_oqit():
    d = pd.read_parquet(P_XUS)
    Xtr, _, ytr, _ = train_test_split(d[XUSUSIYATLAR], d["broker"], test_size=0.25,
                                      stratify=d["broker"], random_state=42)
    m = RandomForestClassifier(n_estimators=150, min_samples_leaf=3,
                               random_state=42, n_jobs=1).fit(Xtr, ytr)
    joblib.dump(m, P_MODEL)


def qadam_baho():
    d = pd.read_parquet(P_XUS)
    _, Xte, _, yte = train_test_split(d[XUSUSIYATLAR], d["broker"], test_size=0.25,
                                      stratify=d["broker"], random_state=42)
    m = joblib.load(P_MODEL)
    p = m.predict(Xte)
    P_BAHO.write_text(json.dumps({
        "f1": round(float(f1_score(yte, p)), 4),
        "aniqlik": round(float(precision_score(yte, p)), 4),
        "toliqlik": round(float(recall_score(yte, p)), 4),
    }, indent=2), encoding="utf-8")


BOSQICHLAR = [
    Bosqich("oqi", qadam_oqi, kirish=(XOM,), chiqish=(str(P_XOM),)),
    Bosqich("tekshir", qadam_tekshir, kirish=(str(P_XOM),), chiqish=(str(P_TOZA),), bogliq=("oqi",)),
    Bosqich("tayyorla", qadam_tayyorla, kirish=(str(P_TOZA),), chiqish=(str(P_XUS),), bogliq=("tekshir",)),
    Bosqich("oqit", qadam_oqit, kirish=(str(P_XUS),), chiqish=(str(P_MODEL),), bogliq=("tayyorla",)),
    Bosqich("baho", qadam_baho, kirish=(str(P_XUS), str(P_MODEL)), chiqish=(str(P_BAHO),), bogliq=("oqit",)),
]

if __name__ == "__main__":
    majburiy = "--majburiy" in sys.argv
    chop_et(ishga_tushir(BOSQICHLAR, majburiy=majburiy))
    if P_BAHO.exists():
        print("\nmetrikalar:", P_BAHO.read_text(encoding="utf-8").replace("\n", " "))

rm -rf oraliq
python3 quvur_ishga.py

python3 quvur_ishga.py

touch oraliq/xususiyat.parquet
python3 quvur_ishga.py

python3 -c "
import sys; sys.path.insert(0, 'src')
from quvur.dag import Bosqich, ishga_tushir, chop_et

def yaxshi(): pass
def yomon(): raise ValueError('ma\'lumot manbasi javob bermadi')

b = [
    Bosqich('oqi', yaxshi),
    Bosqich('tekshir', yomon, bogliq=('oqi',)),
    Bosqich('tayyorla', yaxshi, bogliq=('tekshir',)),
    Bosqich('oqit', yaxshi, bogliq=('tayyorla',)),
    Bosqich('hisobot', yaxshi),
]
chop_et(ishga_tushir(b, holat_yoli='xato_holati.json'))
"

python3 -c "
import sys; sys.path.insert(0, 'src')
from quvur.dag import Bosqich, tartibla

b = [Bosqich('a', lambda: None, bogliq=('b',)),
     Bosqich('b', lambda: None, bogliq=('a',))]
try:
    tartibla(b)
except ValueError as e:
    print('XATO (yaxshi):', e)
"

python3 quvur_ishga.py > /dev/null 2>&1
cat quvur_holati.json

import pandas as pd
show(pd.DataFrame([
    {"asbob": "Airflow", "kuchli tomoni": "Sanoat standarti, boy ekotizim, jadval bo'yicha ishga tushirish",
     "kamchiligi": "Og'ir: bazasi, rejalashtiruvchi, veb-server kerak"},
    {"asbob": "Prefect", "kuchli tomoni": "Python ga yaqin, sodda sintaksis",
     "kamchiligi": "Kichikroq hamjamiyat"},
    {"asbob": "Dagster", "kuchli tomoni": "Ma'lumot aktivlari (assets) markazda, turlar tekshiriladi",
     "kamchiligi": "Boshqacha fikrlash talab qiladi"},
    {"asbob": "Makefile", "kuchli tomoni": "Hech narsa o'rnatish shart emas, keshlash bepul",
     "kamchiligi": "Jadval, qayta urinish, kuzatuv yo'q"},
    {"asbob": "DVC pipelines", "kuchli tomoni": "5-bobdagi ma'lumot versiyalash bilan birlashgan",
     "kamchiligi": "Faqat ML quvurlari uchun"},
]))

from airflow.decorators import dag, task
from datetime import datetime

@dag(schedule="0 3 * * *", start_date=datetime(2026, 9, 1), catchup=False)
def hisobot_quvuri():
    @task
    def oqi(): ...

    @task
    def tekshir(kirish): ...

    @task
    def tayyorla(kirish): ...

    @task
    def oqit(kirish): ...

    # bog'liqlik funksiya chaqiruvi orqali bildiriladi
    tayyorlangan = tayyorla(tekshir(oqi()))
    oqit(tayyorlangan)

hisobot_quvuri()

import pandas as pd
show(pd.DataFrame([
    {"yomon (idempotent emas)": "df.to_csv(yol, mode='a')  # qo'shib yozadi",
     "yaxshi (idempotent)": "df.to_csv(yol)  # ustidan yozadi"},
    {"yomon (idempotent emas)": "INSERT INTO t VALUES (...)",
     "yaxshi (idempotent)": "INSERT ... ON CONFLICT DO UPDATE"},
    {"yomon (idempotent emas)": "hisob += 1",
     "yaxshi (idempotent)": "hisob = manbadan_hisobla()"},
    {"yomon (idempotent emas)": "model_v{n+1}.joblib  # nomeri oshadi",
     "yaxshi (idempotent)": "model_{malumot_hash}.joblib"},
]))