Data Engineering va MLOps: chegara qayerda

diagramma(
    {
        "Ilova bazasi\n(Postgres)": (0, 3.2, 2.3, 1.0, "manba"),
        "Hodisalar\n(klik, buyurtma)": (0, 1.6, 2.3, 1.0, "manba"),
        "Tashqi API\n(karta, ob-havo)": (0, 0.0, 2.3, 1.0, "manba"),
        "Yig'ish\n(Kafka, CDC)": (3.1, 1.6, 2.1, 1.0, "qayta"),
        "Saqlash\n(Iceberg)": (5.9, 1.6, 2.1, 1.0, "saqlash"),
        "Jadval\n(ijara.csv)": (8.7, 1.6, 2.1, 1.0, "saqlash"),
        "Model\n(I va III qism)": (11.5, 1.6, 2.1, 1.0, "xizmat"),
    },
    [
        ("Ilova bazasi\n(Postgres)", "Yig'ish\n(Kafka, CDC)"),
        ("Hodisalar\n(klik, buyurtma)", "Yig'ish\n(Kafka, CDC)"),
        ("Tashqi API\n(karta, ob-havo)", "Yig'ish\n(Kafka, CDC)"),
        ("Yig'ish\n(Kafka, CDC)", "Saqlash\n(Iceberg)"),
        ("Saqlash\n(Iceberg)", "Jadval\n(ijara.csv)"),
        ("Jadval\n(ijara.csv)", "Model\n(I va III qism)"),
    ],
    nom="de_chegara.png", kenglik=15,
    izoh="II qism chapdagi to'rt quti haqida. I va III qism eng o'ngdagisi haqida edi.",
)

python3 -c "
import sqlite3, time, pathlib
import pandas as pd, pyarrow as pa, pyarrow.parquet as pq

d = pd.read_csv('../datasets/buyurtma.csv')
print(f'buyurtmalar: {len(d):,} qator, {len(d.columns)} ustun')

# OLTP uslubi: qatorli saqlash + indeks
con = sqlite3.connect('buyurtma.db')
d.to_sql('buyurtma', con, index=False, if_exists='replace')
con.execute('CREATE INDEX i_kuryer ON buyurtma(kuryer_id)')
con.commit(); con.close()

# OLAP uslubi: ustunli saqlash
pq.write_table(pa.Table.from_pandas(d), 'buyurtma.parquet', compression='zstd')

for f in ['buyurtma.db', 'buyurtma.parquet']:
    print(f'  {f:20s} {pathlib.Path(f).stat().st_size/1024:8.1f} KB')
"

python3 -c "
import sqlite3, time
import numpy as np, pyarrow.parquet as pq

def olch(f, n=20):
    f()                                   # isitish
    v = [ (lambda t0=time.perf_counter(): (f(), (time.perf_counter()-t0)*1000)[1])() for _ in range(n) ]
    return np.median(v)

con = sqlite3.connect('buyurtma.db')

def oltp_sqlite():
    return con.execute(\"SELECT * FROM buyurtma WHERE kuryer_id='K0789'\").fetchall()

def oltp_parquet():
    t = pq.read_table('buyurtma.parquet')
    return t.filter(t['kuryer_id'].to_pandas().values == 'K0789')

def olap_sqlite():
    return con.execute('SELECT tuman, AVG(yol_daq) FROM buyurtma GROUP BY tuman').fetchall()

def olap_parquet():
    t = pq.read_table('buyurtma.parquet', columns=['tuman', 'yol_daq']).to_pandas()
    return t.groupby('tuman')['yol_daq'].mean()

nom1 = 'OLTP: bitta kuryer qatorlari'
nom2 = 'OLAP: tuman kesimida o\'rtacha'
print(f'{\"savol\":34s} {\"SQLite\":>10s} {\"Parquet\":>10s}')
print('-' * 56)
a, b = olch(oltp_sqlite), olch(oltp_parquet)
print(f'{nom1:34s} {a:9.2f}ms {b:9.2f}ms')
c, e = olch(olap_sqlite), olch(olap_parquet)
print(f'{nom2:34s} {c:9.2f}ms {e:9.2f}ms')
print()
print(f'OLTP da SQLite {b/a:.1f}x tez')
print(f'OLAP da Parquet {c/e:.1f}x tez')
con.close()
"

import pandas as pd
show(pd.DataFrame([
    {"": "Batch", "qachon ishlaydi": "Jadval bo'yicha (masalan har tunda)",
     "kechikish": "Soatlar", "misol": "Kunlik broker hisoboti"},
    {"": "Oqim (streaming)", "qachon ishlaydi": "Hodisa kelishi bilan",
     "kechikish": "Soniyalar", "misol": "Ballon skanerlanganda ogohlantirish"},
]))

python3 -c "
import time
import pandas as pd

d = pd.read_csv('../datasets/ballon.csv', parse_dates=['vaqt'])
print(f'jami hodisalar: {len(d):,}')

t0 = time.perf_counter()
natija = d.groupby('ombor').size()
batch_ms = (time.perf_counter() - t0) * 1000

t0 = time.perf_counter()
holat = {}
for omb in d['ombor'].values:
    holat[omb] = holat.get(omb, 0) + 1
oqim_ms = (time.perf_counter() - t0) * 1000

nisbat = oqim_ms / batch_ms
bitta = oqim_ms / len(d) * 1000
oxirgi = d['vaqt'].max()

print()
print(f'BATCH: bitta guruhlash        -> {batch_ms:7.1f} ms')
print(f'OQIM : har hodisada yangilash -> {oqim_ms:7.1f} ms')
print(f'  bitta hodisaga {bitta:.2f} mikrosekund')
print()
print(f'natijalar bir xilmi: {dict(natija) == holat}')
print(f'oqim batchdan {nisbat:.1f}x ko\'p vaqt oldi')
print()
print(f'BATCH javobi {oxirgi} dan keyin tayyor')
print('OQIM  javobi har hodisadan keyin tayyor')
"

import pandas as pd
show(pd.DataFrame([
    {"Vazifa": "Postgres dan o'zgarishlarni olish", "Egasi": "Data Engineer", "Bob": "11"},
    {"Vazifa": "Hodisalar oqimini yig'ish", "Egasi": "Data Engineer", "Bob": "10"},
    {"Vazifa": "Jadval sxemasi va evolyutsiyasi", "Egasi": "Data Engineer", "Bob": "15"},
    {"Vazifa": "Ma'lumot sifati tekshiruvi", "Egasi": "Ikkalasi", "Bob": "6"},
    {"Vazifa": "Xususiyat hisoblash", "Egasi": "Ikkalasi", "Bob": "3"},
    {"Vazifa": "Model o'qitish va baholash", "Egasi": "ML muhandis", "Bob": "24"},
    {"Vazifa": "Model versiyasi va registr", "Egasi": "ML muhandis", "Bob": "23"},
    {"Vazifa": "Inference servis", "Egasi": "ML muhandis", "Bob": "27"},
]))