28.10-dars: Katta ma'lumot bilan ishlash

DARAJA           BELGISI                              ODATIY YECHIM
---------------  -----------------------------------  ----------------------------
1. Xotiraga      jadval RAM ning ~1/3 qismidan kichik  oddiy pandas/numpy
   sig'adi       (amallar nusxa yaratadi: x2-x3 zaxira)
2. Turlar bilan  xom holda sig'maydi, lekin to'g'ri   dtype, category, usecols
   sig'adi       dtype va kerakli ustunlar bilan sig'adi
3. Diskda,       sig'maydi, lekin savol bir o'tishda   chunk, oqimli agregatsiya,
   bir o'tish    javob beradi (yig'indi, sanoq)        sqlite/DuckDB, Parquet
4. Bitta mashina ko'p o'tish yoki og'ir hisob,          ko'p yadroli (Polars,
   sekin         lekin bitta disk yetadi               DuckDB, Dask lokal)
5. Bitta mashina ma'lumot bitta diskka/mashinaga       Spark, Dask klaster,
   yetmaydi      sig'maydi yoki hisob soatlab ketadi   bulut omborlari

QOIDA: pastki darajadagi yechimni sinab ko'rmasdan yuqorisiga o'tmang.
Har daraja yuqoriga - murakkablik, narx va xato manbalari ko'payadi.

TUR        BAYT  DIAPAZON / ANIQLIK
int8        1    -128 .. 127
int16       2    -32768 .. 32767
int32       4    -2.1e9 .. 2.1e9
int64       8    -9.2e18 .. 9.2e18
float32     4    ~7 o'nlik raqam aniqlik; 16 777 216 gacha butunlar ANIQ
float64     8    ~15-16 o'nlik raqam
category    kod (int8/int16) + noyob qiymatlar bir marta
str/object  har qiymat alohida Python satri (~49 bayt + uzunlik)

memory_usage()           - faqat ko'rsatkichlar (object uchun 8 bayt)  -> ALDAYDI
memory_usage(deep=True)  - satrlarning o'zini ham sanaydi               -> HAQIQIY

pd.read_csv(yol, chunksize=50_000)  -> har safar 50 000 qatorli DataFrame
xotirada bir vaqtning o'zida FAQAT bitta bo'lak + yig'uvchi holat

BO'LAKLANADIGAN (birlashtiriladigan) statistikalar:
  sanoq, yig'indi, min, max           -> oddiy qo'shish / min / max
  o'rtacha                            -> (yig'indi, sanoq) juftligi
  dispersiya                          -> (n, o'rtacha, M2) - Welford/Chan
  guruh bo'yicha yig'indi             -> lug'at/Series ni qo'shib borish

BO'LAKLANMAYDIGAN (aniq):
  mediana, kvantillar                 -> bo'laklar medianasidan hosil bo'lmaydi!
  noyob qiymatlar soni                -> barcha qiymatlar to'plami kerak
  tartiblash, rang (rank)             -> global ko'rinish kerak
  YECHIM: taxminiy sketch (HyperLogLog, t-digest) yoki diskda saralash/sqlite

NAIV (bitta o'tish, lekin XAVFLI):
  s = sum x,  s2 = sum x^2
  var = (s2 - s^2 / n) / (n - 1)
  muammo: x ~ 1e9 bo'lsa, s2 ~ n * 1e18 - ikki ULKAN sonning ayirmasi
          float64 aniqligi ~16 raqam -> kichik dispersiya "yutilib ketadi"
          (katastrofik ayirish: natija manfiy ham chiqishi mumkin)

WELFORD (bitta o'tish, BARQAROR):
  n = n + 1
  delta  = x - orta
  orta   = orta + delta / n
  M2     = M2 + delta * (x - orta)       <- yangi orta bilan!
  var    = M2 / (n - 1)
  faqat og'ishlar (kichik sonlar) yig'iladi -> aniqlik saqlanadi

BIRLASHTIRISH (Chan va boshq.) - parallel/map-reduce uchun:
  A = (n_a, orta_a, M2_a),  B = (n_b, orta_b, M2_b)
  n     = n_a + n_b
  delta = orta_b - orta_a
  orta  = orta_a + delta * n_b / n
  M2    = M2_a + M2_b + delta^2 * n_a * n_b / n

birinchi k element -> rezervuarga
i-element (0 dan sanaganda, i >= k):
  j = tasodifiy butun son [0, i] oralig'ida
  agar j < k: rezervuar[j] = element
ISBOT G'OYASI: i-element kiradi ehtimol k/(i+1); keyingi har qadamda
  undan keyin keladigan t-element uni siqib chiqarish ehtimoli 1/(t+1),
  saqlanib qolish (t/(t+1)); ko'paytma teleskopik -> oxirida k/N
  -> HAR BIR element uchun bir xil k/N

HYPERLOGLOG - noyob elementlar soni (cardinality)
  har element -> 64 bitli xesh
  birinchi p bit -> registr raqami (m = 2^p registr)
  qolgan bitlarda boshidagi nollar soni + 1 = rho
  registr = max(registr, rho)      <- takror element hech narsani o'zgartirmaydi
  baho = alfa_m * m^2 / sum 2^(-registr)   (+ kichik diapazon tuzatishi)
  nisbiy xato ~ 1.04 / sqrt(m):   m = 4096 -> ~1.6%, xotira ~4 KB
  birlashtirish: registrlar bo'yicha max

COUNT-MIN SKETCH - chastota (qancha marta uchradi?)
  d qator x w ustun hisoblagich, d ta xesh funksiya
  qo'shish: har qatorda jadval[j][h_j(x)] += 1
  baho:     min_j jadval[j][h_j(x)]
  xato FAQAT yuqoriga (to'qnashuvlar qo'shadi, ayirmaydi): baho >= haqiqiy
  kafolat: baho <= haqiqiy + eps * N   ehtimoli >= 1 - delta
           w = ceil(e / eps),  d = ceil(ln(1 / delta))
  kamyob elementlar uchun NISBIY xato katta (eps*N ular uchun ulkan)

BLOOM FILTER - a'zolik (bu element to'plamda bormi?)
  m bitli massiv, k ta xesh
  qo'shish: k ta bitni 1 qilish;  tekshirish: k tasi ham 1 mi?
  "yo'q"  -> ANIQ yo'q (yolg'on manfiy YO'Q)
  "bor"   -> ehtimol bor (yolg'on musbat ulushi):
             FPR ~ (1 - e^(-k*n/m))^k
  optimal k = (m/n) * ln 2;  element boshiga 10 bit -> FPR ~ 0.8%
  o'chirib bo'lmaydi (counting Bloom filter - hisoblagichlar bilan)

KIRISH: fayl bo'laklarga (split) bo'lingan, har bo'lak - alohida mashinada

1. MAP:     har yozuv -> (kalit, qiymat) juftlari
            "juda yaxshi narx" -> (juda,1) (yaxshi,1) (narx,1)
2. KOMBINATOR (ixtiyoriy, map tomonida lokal reduce):
            (yaxshi,1) (yaxshi,1) (yaxshi,1) -> (yaxshi,3)
            SHART: amal assotsiativ va kommutativ (yig'indi, max, (sum,count))
3. SHUFFLE: har juft kalit bo'yicha reducer ga yuboriladi
            reducer = xesh(kalit) mod R
            TARMOQ orqali - eng qimmat bosqich
4. REDUCE:  bir kalitning barcha qiymatlari -> bitta natija
            (yaxshi, [3, 5, 2]) -> (yaxshi, 10)

MUAMMOLAR:
  shuffle hajmi   - kombinator bilan keskin kamayadi
  issiq kalit     - bitta kalit (masalan "va") bitta reducer ni to'ldiradi
                    (skew); yechim: kalitga tuz qo'shib ikki bosqichli yig'ish
  noto'g'ri kombinator - o'rtachalar o'rtachasi 2.3-bob
  deterministik bo'lmagan xesh - Python hash() jarayonlar orasida farq qiladi

INDEKSSIZ:   WHERE mijoz_id = 42   -> SCAN savdo (hamma qator o'qiladi)  O(N)
INDEKS BILAN: B-daraxt (mijoz_id -> qator joyi)
             WHERE mijoz_id = 42   -> SEARCH ... USING INDEX            O(log N + k)

EXPLAIN QUERY PLAN <so'rov>  - baza qanday bajarishini ko'rsatadi:
  SCAN jadval                          - to'liq o'qish
  SEARCH jadval USING INDEX i (x=?)     - indeks bo'yicha qidiruv
  USING COVERING INDEX                  - jadvalga umuman qaytmaydi

INDEKS NARXI:
  disk joyi (qo'shimcha B-daraxt), har INSERT/UPDATE sekinroq
  past selektivlik (qatorlarning ko'pchiligi mos) - indeks FOYDASIZ
  yoki hatto zararli: har qator uchun indeksdan jadvalga "sakrash"
KOMPOZIT INDEKS (dokon, kun): chap prefiks qoidasi -
  WHERE dokon = ? AND kun BETWEEN ? AND ?  - ishlaydi
  WHERE kun BETWEEN ? AND ?                - dokon siz ishlamaydi

PARQUET:
  ustunli    -> faqat kerakli ustunlar o'qiladi (projection pushdown)
  siqilgan   -> bir turdagi qiymatlar yonma-yon: dictionary, RLE, snappy/zstd
                (CSV ga nisbatan odatda 5-10 barobar kichik)
  sxema bor  -> turlar faylda saqlanadi, har o'qishda qayta taxmin qilinmaydi
  row group statistikasi (min/max) -> WHERE kun > 300 bo'lsa, mos kelmaydigan
                bloklar umuman o'qilmaydi (predicate pushdown)
  bo'limlash (partitioning): savdo/yil=2025/oy=03/part-0.parquet
                -> faqat kerakli papkalar o'qiladi

SPARK (Apache Spark, PySpark)
  klaster: driver (reja tuzadi) + executor lar (bo'limlarni qayta ishlaydi)
  RDD        - past darajali, taqsimlangan kolleksiya (map, reduceByKey)
  DataFrame  - sxemali jadval; Catalyst optimallashtiruvchisi rejani qayta yozadi
  LAZY       - transformatsiyalar (select, filter, groupBy) faqat REJA quradi;
               action (count, collect, write) chaqirilganda bajariladi
  bo'limlar (partitions) - parallellik birligi; juda kam -> yadrolar bo'sh,
               juda ko'p -> vazifalar boshqaruvi xarajati
  narrow transformatsiya (map, filter) - shuffle siz
  wide transformatsiya (groupBy, join) - SHUFFLE, bosqich chegarasi
  cache/persist - bir natija ko'p marta ishlatilsa

DASK       - pandas/numpy API sini bo'laklarga bo'lib, vazifalar grafiga
             aylantiradi; lokal ko'p yadroda yoki klasterda; lazy (.compute())
POLARS     - Rust da yozilgan, ustunli (Arrow), ko'p yadroli DataFrame;
             lazy rejim (scan_csv/scan_parquet -> collect) so'rovni optimallaydi;
             bitta mashinada pandas dan ko'pincha ancha tez
DUCKDB     - jarayon ichidagi (sqlite kabi) tahliliy SQL baza; ustunli,
             vektorlashgan; Parquet/CSV ni to'g'ridan-to'g'ri so'raydi;
             xotiradan katta ma'lumotda diskka to'kib (spill) ishlaydi

AVVAL SHULARNI SINANG:
  1. kerakli ustunlar + to'g'ri dtype   -> ko'pincha 3-10 barobar kichik
  2. filtr erta (faqat kerakli davr/hudud)
  3. chunk bilan bir o'tish (yig'indi, sanoq, Welford)
  4. namuna: model prototipi uchun 1-5% tasodifiy (yoki qatlamli) namuna
  5. sqlite/DuckDB + Parquet - bitta faylda SQL
  6. bitta KATTA mashina (bulutda 256 GB RAM - soatiga bir necha dollar)

TAQSIMLANGAN VOSITA KERAK BO'LADI:
  ma'lumot bitta mashina diskiga sig'maydi yoki doimiy o'sib boradi
  og'ir join/groupBy ko'p terabaytda, muntazam (har kuni)
  jamoa allaqachon klaster va quvurlarga ega (Databricks, EMR)

NARXI: klaster boshqaruvi, serializatsiya, shuffle, qiyin debug,
       "kichik ma'lumotda sekinroq" (ishga tushirish xarajati)

import numpy as np
import pandas as pd

# xotira
df.memory_usage(deep=True, index=False).sum()           # HAQIQIY bayt
df["miqdor"] = pd.to_numeric(df["miqdor"], downcast="integer")
df["narx"] = df["narx"].astype(np.float32)
df["dokon"] = df["dokon"].astype("category")            # noyob qiymat kam bo'lsa

# o'qishda turlar va ustunlar
turlar = {"dokon": "category", "miqdor": "int8", "narx": "float32"}
df = pd.read_csv(yol, usecols=list(turlar), dtype=turlar)

# chunk bo'yicha oqimli agregatsiya
jami, soni, guruh = 0.0, 0, pd.Series(dtype=float)
for bolak in pd.read_csv(yol, usecols=["dokon", "summa"], chunksize=100_000):
    jami += bolak["summa"].sum()
    soni += len(bolak)
    guruh = guruh.add(bolak.groupby("dokon")["summa"].sum(), fill_value=0)
ortacha = jami / soni                                    # o'rtachalar o'rtachasi EMAS


# Welford
def welford_qosh(holat, x):
    n, orta, m2 = holat
    n += 1
    d = x - orta
    orta += d / n
    m2 += d * (x - orta)
    return n, orta, m2


# sqlite
# db.execute("CREATE INDEX idx_mijoz ON savdo(mijoz_id)")
# db.execute("EXPLAIN QUERY PLAN SELECT ... WHERE mijoz_id = ?", (42,)).fetchall()

# PySpark - lazy: select/filter/groupBy reja quradi, write/collect bajaradi
from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("savdo").getOrCreate()
df = spark.read.parquet("s3://savdo/yil=2025/")          # bo'limlangan Parquet
natija = (df.filter(F.col("kun") >= 300)                 # predicate pushdown
            .groupBy("dokon", "guruh")                   # wide -> shuffle
            .agg(F.sum("summa").alias("tushum"),
                 F.approx_count_distinct("mijoz_id").alias("mijozlar")))
natija.write.mode("overwrite").parquet("s3://hisobot/tushum/")

# Dask - pandas API, bo'laklar, lazy
import dask.dataframe as dd
ddf = dd.read_csv("savdo-*.csv", dtype={"dokon": "category"})
tushum = ddf.groupby("dokon")["summa"].sum().compute()

# Polars - lazy so'rov, ko'p yadroli
import polars as pl
tushum = (pl.scan_parquet("savdo/*.parquet")
            .filter(pl.col("kun") >= 300)
            .group_by("dokon").agg(pl.col("summa").sum())
            .collect())

# DuckDB - fayl ustida to'g'ridan-to'g'ri SQL
import duckdb
tushum = duckdb.sql("""
    SELECT dokon, SUM(summa) AS tushum
    FROM 'savdo/*.parquet' WHERE kun >= 300 GROUP BY dokon
""").df()

# Parquet (pyarrow yoki fastparquet kerak)
df.to_parquet("savdo.parquet", index=False)
qism = pd.read_parquet("savdo.parquet", columns=["dokon", "summa"])

katta = resursga nisbatan; avval dtype, usecols, filtr, namuna
chunk: faqat birlashtiriladigan holat (sum, count, Welford, sketch)
mediana/noyob soni - bo'laklanmaydi -> sketch yoki disk
sketch: kichik xotira, isbotlangan xato, birlashtiriladi
map-reduce: map -> (kombinator) -> shuffle -> reduce; shuffle eng qimmat
indeks: selektiv so'rovga; EXPLAIN QUERY PLAN bilan tekshir

"""Xotirani tejash: dtype optimallash, CSV ni chunk bo'yicha o'qish va oqimli agregatsiya."""

import tempfile
from pathlib import Path

import numpy as np
import pandas as pd

SHAHARLAR = ["Toshkent", "Samarqand", "Buxoro", "Namangan", "Andijon"]
DOKONLAR = [f"{s}-{i:02d}" for s in SHAHARLAR for i in range(1, 5)]
GURUHLAR = ["non", "sut", "gosht", "meva", "ichimlik", "maishiy"]
TOLOV = ["naqd", "karta", "QR"]


def yarat(n, seed=0):
    """Sintetik cheklar jadvali (bir yil, 20 do'kon)."""
    rng = np.random.default_rng(seed)
    kun = np.sort(rng.integers(1, 366, n))          # fayl vaqt tartibida
    return pd.DataFrame({
        "chek_id": np.arange(n, dtype=np.int64) + 50_000_000,
        "kun": kun,
        "dokon": rng.choice(DOKONLAR, n),
        "guruh": rng.choice(GURUHLAR, n, p=[0.25, 0.2, 0.1, 0.15, 0.2, 0.1]),
        "miqdor": rng.integers(1, 13, n),
        "narx": np.round(rng.lognormal(9.3, 0.7, n)            # so'm, yil
                         * (1 + 0.5 * (kun / 365) ** 2), -2),   # oxiriga qimmat
        "chegirma": rng.choice([0.0, 0.05, 0.1, 0.2], n,
                               p=[0.7, 0.15, 0.1, 0.05]),
        "tolov": rng.choice(TOLOV, n, p=[0.35, 0.5, 0.15]),
    })


def optimallash(df):
    """Butun -> eng kichik butun, kasr -> float32, kam noyob matn -> category."""
    natija = df.copy()
    for ustun in natija.columns:
        s = natija[ustun]
        if pd.api.types.is_integer_dtype(s):
            natija[ustun] = pd.to_numeric(s, downcast="integer")
        elif pd.api.types.is_float_dtype(s):
            natija[ustun] = s.astype(np.float32)
        elif s.nunique() / len(s) < 0.5:
            natija[ustun] = s.astype("category")
    return natija


def tushum(df):
    return (df["miqdor"].astype(np.float64) * df["narx"].astype(np.float64)
            * (1 - df["chegirma"].astype(np.float64)))


def main() -> None:
    n = 400_000
    df = yarat(n)
    opt = optimallash(df)

    print("=== 1. Xotira: ustunlar bo'yicha (memory_usage(deep=True)) ===")
    oldin = df.memory_usage(deep=True, index=False)
    keyin = opt.memory_usage(deep=True, index=False)
    sayoz = df.memory_usage(deep=False, index=False)
    print(f"  {'ustun':<9} {'tur':>8} {'bayt':>11} {'yangi tur':>10} "
          f"{'bayt':>10}")
    for u in df.columns:
        print(f"  {u:<9} {str(df[u].dtype):>8} {oldin[u]:>11,} "
              f"{str(opt[u].dtype):>10} {keyin[u]:>10,}")
    print(f"  jami: {oldin.sum():,} -> {keyin.sum():,} bayt "
          f"({oldin.sum() / keyin.sum():.1f} barobar kichik)")
    print(f"  deep=False (aldaydi): {sayoz.sum():,} bayt")

    print("\n=== 2. Optimallash to'g'riligini tekshirish ===")
    t64 = tushum(df).sum()
    t32 = (opt["miqdor"] * opt["narx"] * (1 - opt["chegirma"])).sum()
    print(f"  tushum float64: {t64:,.0f} so'm")
    print(f"  tushum float32: {float(t32):,.0f} so'm "
          f"(nisbiy farq {abs(float(t32) - t64) / t64:.1e})")
    print(f"  narx float32 da aniqmi: "
          f"{bool((opt['narx'].astype(np.float64) == df['narx']).all())}")
    print(f"  miqdor yig'indisi: int64 {int(df['miqdor'].sum()):,}, "
          f"int8 {int(opt['miqdor'].sum()):,}")
    toldi = opt["miqdor"] * 20
    print(f"  TUZOQ: int8 miqdor * 20 -> max {int(toldi.max())}, "
          f"min {int(toldi.min())} (tur {toldi.dtype})")
    kod = "CHK" + df["chek_id"].astype(str)
    print(f"  noyob matn (chek kodi): str {kod.memory_usage(deep=True, index=False):,}"
          f" bayt, category "
          f"{kod.astype('category').memory_usage(deep=True, index=False):,} bayt")

    turlar = {"chek_id": "int32", "kun": "int16", "dokon": "category",
              "guruh": "category", "miqdor": "int8", "narx": "float32",
              "chegirma": "float32", "tolov": "category"}
    kerakli = ["dokon", "guruh", "miqdor", "narx", "chegirma"]
    oqim_turlar = {"miqdor": "int64", "narx": "float64", "chegirma": "float64"}
    with tempfile.TemporaryDirectory() as papka:
        yol = Path(papka) / "cheklar.csv"
        df.to_csv(yol, index=False)

        print("\n=== 3. CSV ni o'qish: standart, turlar bilan, usecols ===")
        print(f"  fayl hajmi: {yol.stat().st_size:,} bayt")
        a = pd.read_csv(yol)
        b = pd.read_csv(yol, dtype=turlar)
        c = pd.read_csv(yol, usecols=kerakli,
                        dtype={u: turlar[u] for u in kerakli})
        for nom, d in [("standart", a), ("dtype bilan", b),
                       ("dtype + usecols", c)]:
            print(f"  {nom:<16} {d.memory_usage(deep=True, index=False).sum():>12,}"
                  f" bayt, {d.shape[1]} ustun")

        print("\n=== 4. Chunk bo'yicha oqimli agregatsiya (50 000 qator) ===")
        jami, soni, bolaklar, maks_bayt = 0.0, 0, 0, 0
        dokon_sum = pd.Series(dtype=np.float64)
        guruh_sum = pd.Series(dtype=np.float64)
        guruh_n = pd.Series(dtype=np.float64)
        bolak_med, bolak_p99 = [], []
        for bolak in pd.read_csv(yol, usecols=kerakli, dtype=oqim_turlar,
                                 chunksize=50_000):
            t = tushum(bolak)
            jami += t.sum()
            soni += len(t)
            bolaklar += 1
            maks_bayt = max(maks_bayt,
                            int(bolak.memory_usage(deep=True, index=False).sum()))
            dokon_sum = dokon_sum.add(t.groupby(bolak["dokon"]).sum(),
                                      fill_value=0)
            guruh_sum = guruh_sum.add(t.groupby(bolak["guruh"]).sum(),
                                      fill_value=0)
            guruh_n = guruh_n.add(t.groupby(bolak["guruh"]).size(),
                                  fill_value=0)
            bolak_med.append(float(t.median()))
            bolak_p99.append(float(t.quantile(0.99)))
        toliq = pd.read_csv(yol, usecols=kerakli, dtype=oqim_turlar)
        tt = tushum(toliq)
        print(f"  bo'laklar: {bolaklar}, bitta bo'lak maks {maks_bayt:,} bayt")
        print(f"  o'rtacha chek: oqim {jami / soni:,.4f}, to'liq {tt.mean():,.4f}")
        d_tol = tt.groupby(toliq["dokon"]).sum()
        g_tol = tt.groupby(toliq["guruh"]).mean()
        d_farq = float((dokon_sum - d_tol).abs().max() / d_tol.abs().max())
        g_farq = float(((guruh_sum / guruh_n) - g_tol).abs().max())
        print(f"  do'kon yig'indilari: maks nisbiy farq {d_farq:.1e}")
        print(f"  guruh o'rtachalari:  maks farq {g_farq:.1e} so'm")
        mos = (np.isclose(jami / soni, tt.mean(), rtol=1e-12)
               and d_farq < 1e-12 and g_farq < 1e-6)
        print(f"  oqim natijasi to'liq o'qish bilan mos: {mos}")
        print(f"  mediana: to'liq {tt.median():,.1f}; bo'lak medianalari: "
              f"o'rtachasi {np.mean(bolak_med):,.1f}, medianasi "
              f"{np.median(bolak_med):,.1f}")
        print(f"  99-persentil: to'liq {tt.quantile(0.99):,.1f}; bo'laklar "
              f"o'rtachasi {np.mean(bolak_p99):,.1f}")
        eng = d_tol.sort_values(ascending=False)
        print(f"  eng katta tushum: {eng.index[0]} {eng.iloc[0]:,.0f}, "
              f"eng kichik: {eng.index[-1]} {eng.iloc[-1]:,.0f}")
    print("  ⭐ Bo'laklarda faqat birlashtiriladigan holat: yig'indi, sanoq")


if __name__ == "__main__":
    main()

=== 1. Xotira: ustunlar bo'yicha (memory_usage(deep=True)) ===
  ustun          tur        bayt  yangi tur       bayt
  chek_id      int64   3,200,000      int32  1,600,000
  kun          int64   3,200,000      int16    800,000
  dokon          str  23,838,808   category    401,192
  guruh          str  21,501,618   category    400,324
  miqdor       int64   3,200,000       int8    400,000
  narx       float64   3,200,000    float32  1,600,000
  chegirma   float64   3,200,000    float32  1,600,000
  tolov          str  21,279,529   category    400,158
  jami: 82,619,955 -> 7,201,674 bayt (11.5 barobar kichik)
  deep=False (aldaydi): 25,600,000 bayt

=== 2. Optimallash to'g'riligini tekshirish ===
  tushum float64: 41,160,657,890 so'm
  tushum float32: 41,160,658,944 so'm (nisbiy farq 2.6e-08)
  narx float32 da aniqmi: True
  miqdor yig'indisi: int64 2,598,836, int8 2,598,836
  TUZOQ: int8 miqdor * 20 -> max 120, min -116 (tur int8)
  noyob matn (chek kodi): str 24,000,000 bayt, category 25,600,000 bayt

=== 3. CSV ni o'qish: standart, turlar bilan, usecols ===
  fayl hajmi: 20,115,922 bayt
  standart           82,619,955 bayt, 8 ustun
  dtype bilan         7,202,510 bayt, 8 ustun
  dtype + usecols     4,402,244 bayt, 5 ustun

=== 4. Chunk bo'yicha oqimli agregatsiya (50 000 qator) ===
  bo'laklar: 8, bitta bo'lak maks 6,868,073 bayt
  o'rtacha chek: oqim 102,901.6447, to'liq 102,901.6447
  do'kon yig'indilari: maks nisbiy farq 0.0e+00
  guruh o'rtachalari:  maks farq 0.0e+00 so'm
  oqim natijasi to'liq o'qish bilan mos: True
  mediana: to'liq 70,200.0; bo'lak medianalari: o'rtachasi 70,833.1, medianasi 68,700.0
  99-persentil: to'liq 529,204.0; bo'laklar o'rtachasi 521,852.7
  eng katta tushum: Andijon-04 2,088,367,005, eng kichik: Andijon-03 2,030,503,430
  ⭐ Bo'laklarda faqat birlashtiriladigan holat: yig'indi, sanoq

"""Bir o'tishli algoritmlar: Welford (va Chan birlashtirishi), rezervuar namuna."""

import random

import numpy as np
from scipy import stats


class Welford:
    """O'rtacha va dispersiya - bitta o'tishda, O(1) xotira."""

    def __init__(self):
        self.n, self.orta, self.m2 = 0, 0.0, 0.0

    def qosh(self, x):
        self.n += 1
        delta = x - self.orta
        self.orta += delta / self.n
        self.m2 += delta * (x - self.orta)

    @property
    def dispersiya(self):
        return self.m2 / (self.n - 1)


def birlashtir(a, b):
    """Chan va boshq.: ikki (n, orta, m2) holatini birlashtirish."""
    n = a[0] + b[0]
    delta = b[1] - a[1]
    orta = a[1] + delta * b[0] / n
    m2 = a[2] + b[2] + delta * delta * a[0] * b[0] / n
    return n, orta, m2


def naiv(xs):
    """sum x va sum x^2 - bitta o'tish, lekin katastrofik ayirish."""
    n, s, s2 = 0, 0.0, 0.0
    for x in xs:
        n += 1
        s += x
        s2 += x * x
    return s / n, (s2 - s * s / n) / (n - 1)


def rezervuar(oqim, k, rnd):
    """Algorithm R: har element k/N ehtimol bilan."""
    namuna = []
    for i, x in enumerate(oqim):
        if i < k:
            namuna.append(x)
        else:
            j = rnd.randrange(i + 1)
            if j < k:
                namuna[j] = x
    return namuna


def yangiga_moyil(oqim, k, rnd):
    """NOTO'G'RI: har yangi element 1/2 ehtimol bilan kiradi."""
    namuna = []
    for i, x in enumerate(oqim):
        if i < k:
            namuna.append(x)
        elif rnd.random() < 0.5:
            namuna[rnd.randrange(k)] = x
    return namuna


def main() -> None:
    rng = np.random.default_rng(0)
    n = 200_000
    shovqin = rng.normal(0, 50, n)                 # std 50 so'm

    print("=== 1. Dispersiya: naiv formula va Welford (haqiqiy var ~2500) ===")
    print(f"  {'siljish':>8} {'ikki o_tish':>12} {'naiv':>14} {'Welford':>12} "
          f"{'naiv xato':>10} {'W xato':>9}")
    for siljish in [0.0, 1e6, 1e8, 1e9]:
        x = siljish + shovqin
        xs = x.tolist()
        ikki = float(np.var(x, ddof=1))
        _, v_naiv = naiv(xs)
        w = Welford()
        for v in xs:
            w.qosh(v)
        print(f"  {siljish:>8.0e} {ikki:>12.4f} {v_naiv:>14.4f} "
              f"{w.dispersiya:>12.4f} {abs(v_naiv - ikki) / ikki:>10.1e} "
              f"{abs(w.dispersiya - ikki) / ikki:>9.1e}")

    print("\n=== 2. Bo'laklar holatini birlashtirish (Chan) ===")
    x = 1e9 + shovqin
    bolaklar = np.array_split(x, 8)
    holatlar = []
    for b in bolaklar:
        orta = float(b.mean())
        holatlar.append((len(b), orta, float(np.sum((b - orta) ** 2))))
    jami = holatlar[0]
    for h in holatlar[1:]:
        jami = birlashtir(jami, h)
    ikki = float(np.var(x, ddof=1))
    print(f"  8 bo'lak, hajmlar {[len(b) for b in bolaklar][:3]} (+5 ta)")
    print(f"  birlashtirilgan var {jami[2] / (jami[0] - 1):.6f}, "
          f"to'liq ikki o'tish {ikki:.6f}")
    print(f"  o'rtacha: {jami[1]:.6f} vs {float(x.mean()):.6f}")
    print(f"  mos: {abs(jami[2] / (jami[0] - 1) - ikki) / ikki < 1e-9}")

    print("\n=== 3. Rezervuar namuna: bir xillik testi ===")
    N, k, takror = 200, 10, 4000
    rnd = random.Random(1)
    sanoq_r = np.zeros(N)
    sanoq_y = np.zeros(N)
    for _ in range(takror):
        sanoq_r[rezervuar(range(N), k, rnd)] += 1
        sanoq_y[yangiga_moyil(range(N), k, rnd)] += 1
    kutilgan = takror * k / N
    print(f"  oqim N={N}, namuna k={k}, {takror} takror; "
          f"har pozitsiya kutilgan {kutilgan:.0f} marta")
    print(f"  {'usul':<16} {'1-50':>6} {'51-100':>7} {'101-150':>8} "
          f"{'151-200':>8} {'chi^2':>9} {'p':>9}")
    for nom, s in [("rezervuar (R)", sanoq_r), ("yangiga moyil", sanoq_y)]:
        chorak = [s[i:i + 50].sum() / s.sum() for i in range(0, N, 50)]
        r = stats.chisquare(s)
        print(f"  {nom:<16} {chorak[0]:>6.3f} {chorak[1]:>7.3f} "
              f"{chorak[2]:>8.3f} {chorak[3]:>8.3f} {r.statistic:>9.1f} "
              f"{r.pvalue:>9.2e}")
    for nom, s in [("rezervuar (R)", sanoq_r), ("yangiga moyil", sanoq_y)]:
        bir_xil = stats.chisquare(s).pvalue > 0.01
        print(f"  {nom}: bir xil taqsimot rad etilmadi: {bir_xil}")

    print("\n=== 4. Amaliy: 300 000 tranzaksiyadan 2 000 lik namuna ===")
    summalar = np.round(rng.lognormal(11.5, 1.0, 300_000), -2)
    namuna = np.array(rezervuar(summalar.tolist(), 2000, random.Random(7)))
    se = namuna.std(ddof=1) / np.sqrt(len(namuna))
    print(f"  haqiqiy o'rtacha {summalar.mean():,.0f}, namuna "
          f"{namuna.mean():,.0f} +- {2 * se:,.0f} (2*SE)")
    print(f"  haqiqiy mediana {np.median(summalar):,.0f}, namuna "
          f"{np.median(namuna):,.0f}")
    ichida = abs(namuna.mean() - summalar.mean()) <= 2 * se
    print(f"  haqiqiy o'rtacha 2*SE oralig'ida: {ichida}")
    print(f"  xotira: {len(namuna)} element (oqim {len(summalar):,})")
    print("  ⭐ Oqimda: Welford - barqaror; rezervuar - haqiqatan tasodifiy")


if __name__ == "__main__":
    main()

=== 1. Dispersiya: naiv formula va Welford (haqiqiy var ~2500) ===
   siljish  ikki o_tish           naiv      Welford  naiv xato    W xato
     0e+00    2506.0158      2506.0158    2506.0158    2.3e-14   4.2e-15
     1e+06    2506.0158      2506.0330    2506.0158    6.9e-06   3.2e-14
     1e+08    2506.0158      2293.7715    2506.0158    8.5e-02   1.0e-10
     1e+09    2506.0158     17616.1649    2506.0158    6.0e+00   1.5e-10

=== 2. Bo'laklar holatini birlashtirish (Chan) ===
  8 bo'lak, hajmlar [25000, 25000, 25000] (+5 ta)
  birlashtirilgan var 2506.015807, to'liq ikki o'tish 2506.015807
  o'rtacha: 1000000000.006534 vs 1000000000.006534
  mos: True

=== 3. Rezervuar namuna: bir xillik testi ===
  oqim N=200, namuna k=10, 4000 takror; har pozitsiya kutilgan 200 marta
  usul               1-50  51-100  101-150  151-200     chi^2         p
  rezervuar (R)     0.252   0.250    0.248    0.250     218.2  1.66e-01
  yangiga moyil     0.000   0.006    0.071    0.923  165435.3  0.00e+00
  rezervuar (R): bir xil taqsimot rad etilmadi: True
  yangiga moyil: bir xil taqsimot rad etilmadi: False

=== 4. Amaliy: 300 000 tranzaksiyadan 2 000 lik namuna ===
  haqiqiy o'rtacha 163,360, namuna 171,358 +- 9,968 (2*SE)
  haqiqiy mediana 98,900, namuna 100,900
  haqiqiy o'rtacha 2*SE oralig'ida: True
  xotira: 2000 element (oqim 300,000)
  ⭐ Oqimda: Welford - barqaror; rezervuar - haqiqatan tasodifiy

"""Sketch lar noldan: HyperLogLog, Count-Min Sketch, Bloom filter (numpy bilan)."""

import math
import sys

import numpy as np

MASKA = (1 << 64) - 1


def xesh(x, tuz=0):
    """splitmix64: butun sonlar massivi -> 64 bitli xesh (deterministik)."""
    z = x.astype(np.uint64) + np.uint64((0x9E3779B97F4A7C15 * (tuz + 1)) & MASKA)
    z = (z ^ (z >> np.uint64(30))) * np.uint64(0xBF58476D1CE4E5B9)
    z = (z ^ (z >> np.uint64(27))) * np.uint64(0x94D049BB133111EB)
    return z ^ (z >> np.uint64(31))


def bit_uzunligi(x):
    """Har element uchun int.bit_length() - vektorlashgan, aniq."""
    x = x.copy()
    n = np.zeros(x.shape, dtype=np.int64)
    for s in (32, 16, 8, 4, 2, 1):
        katta = x >= (np.uint64(1) << np.uint64(s))
        n += np.where(katta, s, 0)
        x = np.where(katta, x >> np.uint64(s), x)
    return n + (x > 0)


class HyperLogLog:
    def __init__(self, p, tuz=0):
        self.p, self.m, self.tuz = p, 1 << p, tuz
        self.reg = np.zeros(self.m, dtype=np.uint8)

    def qosh(self, ids):
        h = xesh(ids, self.tuz)
        idx = (h >> np.uint64(64 - self.p)).astype(np.int64)
        qolgan = h & np.uint64((1 << (64 - self.p)) - 1)
        rho = (64 - self.p) - bit_uzunligi(qolgan) + 1
        np.maximum.at(self.reg, idx, rho.astype(np.uint8))

    def baho(self):
        m = self.m
        alfa = 0.7213 / (1 + 1.079 / m)
        e = alfa * m * m / float(np.sum(2.0 ** -self.reg.astype(np.float64)))
        bosh = int(np.count_nonzero(self.reg == 0))
        if e <= 2.5 * m and bosh > 0:            # kichik diapazon tuzatishi
            e = m * math.log(m / bosh)
        return e

    def birlashtir(self, boshqa):
        yangi = HyperLogLog(self.p, self.tuz)
        yangi.reg = np.maximum(self.reg, boshqa.reg)
        return yangi


class CountMin:
    def __init__(self, w, d, tuz=100):
        self.w, self.d, self.tuz = w, d, tuz
        self.jadval = np.zeros((d, w), dtype=np.int64)

    def _ustun(self, ids, j):
        return (xesh(ids, self.tuz + j) % np.uint64(self.w)).astype(np.int64)

    def qosh(self, ids):
        for j in range(self.d):
            self.jadval[j] += np.bincount(self._ustun(ids, j), minlength=self.w)

    def baho(self, ids):
        return np.min([self.jadval[j][self._ustun(ids, j)]
                       for j in range(self.d)], axis=0)


class Bloom:
    def __init__(self, m, k):
        self.m, self.k = m, k
        self.bit = np.zeros(m, dtype=bool)

    def _joylar(self, ids):
        h1 = xesh(ids, 7)
        h2 = xesh(ids, 8) | np.uint64(1)
        return [((h1 + np.uint64(i) * h2) % np.uint64(self.m)).astype(np.int64)
                for i in range(self.k)]

    def qosh(self, ids):
        for j in self._joylar(ids):
            self.bit[j] = True

    def bormi(self, ids):
        natija = np.ones(len(ids), dtype=bool)
        for j in self._joylar(ids):
            natija &= self.bit[j]
        return natija


def main() -> None:
    rng = np.random.default_rng(0)

    print("=== 1. HyperLogLog: kunlik noyob foydalanuvchilar ===")
    faollik = rng.lognormal(0, 1.5, 600_000)
    ehtimol = faollik / faollik.sum()
    kun1 = rng.choice(600_000, 1_500_000, p=ehtimol)
    kun2 = rng.choice(600_000, 1_500_000, p=ehtimol) + 200_000
    aniq1 = len(np.unique(kun1))
    toplam = set(kun1.tolist())
    set_bayt = sys.getsizeof(toplam) + sum(sys.getsizeof(v) for v in toplam)
    print(f"  hodisalar {len(kun1):,}, aniq noyob {aniq1:,}")
    print(f"  aniq (Python set): {set_bayt:,} bayt")
    print(f"  {'p':>3} {'registr':>8} {'bayt':>7} {'baho':>10} "
          f"{'xato':>7} {'nazariya':>9}")
    for p in [8, 10, 12, 14]:
        hll = HyperLogLog(p)
        hll.qosh(kun1)
        e = hll.baho()
        print(f"  {p:>3} {hll.m:>8,} {hll.m:>7,} {e:>10,.0f} "
              f"{(e - aniq1) / aniq1:>+7.2%} {1.04 / math.sqrt(hll.m):>9.2%}")

    print("\n=== 2. HLL xatosi: 30 xil xesh tuzi bo'yicha (p=10 va 12) ===")
    noyob = np.unique(kun1)          # takror registrni o'zgartirmaydi
    for p in [10, 12]:
        xatolar = []
        for tuz in range(30):
            hll = HyperLogLog(p, tuz)
            hll.qosh(noyob)
            xatolar.append((hll.baho() - aniq1) / aniq1)
        xatolar = np.array(xatolar)
        print(f"  p={p}: o'rtacha xato {xatolar.mean():+.3%}, RMS "
              f"{np.sqrt(np.mean(xatolar ** 2)):.3%}, nazariya "
              f"{1.04 / math.sqrt(1 << p):.3%}")
    h1, h2 = HyperLogLog(12), HyperLogLog(12)
    h1.qosh(kun1)
    h2.qosh(kun2)
    aniq_birl = len(np.union1d(kun1, kun2))
    e = h1.birlashtir(h2).baho()
    print(f"  ikki kun birlashmasi: aniq {aniq_birl:,}, HLL (max registr) "
          f"{e:,.0f}, xato {(e - aniq_birl) / aniq_birl:+.2%}")

    print("\n=== 3. Count-Min Sketch: mahsulot ko'rishlari ===")
    n_mah = 100_000
    zipf = 1 / np.arange(1, n_mah + 1) ** 1.1
    korish = rng.choice(n_mah, 2_000_000, p=zipf / zipf.sum())
    haqiqiy = np.bincount(korish, minlength=n_mah)
    bor = np.flatnonzero(haqiqiy > 0)
    N = len(korish)
    print(f"  ko'rishlar N={N:,}, ko'rilgan mahsulotlar {len(bor):,}")
    print(f"  {'eps':>7} {'d':>2} {'w':>6} {'bayt':>9} {'min xato':>8} "
          f"{'<=eps*N':>8} {'o_rt xato':>9} {'kamyob nisbiy':>13}")
    for eps, d in [(0.001, 1), (0.001, 4), (0.0001, 4)]:
        w = math.ceil(math.e / eps)
        cms = CountMin(w, d)
        cms.qosh(korish)
        b = cms.baho(bor)
        xato = b - haqiqiy[bor]
        kamyob = haqiqiy[bor] <= 5
        nisbiy = float(np.mean(xato[kamyob] / haqiqiy[bor][kamyob]))
        print(f"  {eps:>7} {d:>2} {w:>6,} {d * w * 4:>9,} {int(xato.min()):>8} "
              f"{np.mean(xato <= eps * N):>8.3f} {xato.mean():>9.1f} "
              f"{nisbiy:>13.1f}")
    top_aniq = set(bor[np.argsort(-haqiqiy[bor], kind="stable")[:10]].tolist())
    top_cms = set(bor[np.argsort(-b, kind="stable")[:10]].tolist())
    print(f"  top-10 mos kelishi (eps=0.0001, d=4): {len(top_aniq & top_cms)}/10")
    print(f"  aniq lug'at: {len(bor):,} kalit x (8+8) bayt ~ "
          f"{len(bor) * 16:,} bayt (kamida)")

    print("\n=== 4. Bloom filter: qora ro'yxatdagi telefonlar ===")
    telefonlar = rng.choice(np.arange(998_900_000_000, 998_999_999_999),
                            400_000, replace=False)
    royxat, begona = telefonlar[:100_000], telefonlar[100_000:]
    n = len(royxat)
    print(f"  ro'yxatda n={n:,}, tekshiriladigan begona {len(begona):,}")
    print(f"  {'bit/el':>6} {'k':>3} {'bayt':>9} {'FPR o_lchangan':>15} "
          f"{'FPR nazariya':>13} {'yolg_on manfiy':>15}")
    for b_el, k in [(4, 3), (8, 6), (10, 7), (16, 11), (8, 1), (8, 12)]:
        m = b_el * n
        bf = Bloom(m, k)
        bf.qosh(royxat)
        fpr = float(np.mean(bf.bormi(begona)))
        nazariya = (1 - math.exp(-k * n / m)) ** k
        yolgon_manfiy = int(np.sum(~bf.bormi(royxat)))
        print(f"  {b_el:>6} {k:>3} {m // 8:>9,} {fpr:>15.4f} "
              f"{nazariya:>13.4f} {yolgon_manfiy:>15}")
    print(f"  optimal k = (m/n) ln2: 8 bit -> {8 * math.log(2):.1f}, "
          f"10 bit -> {10 * math.log(2):.1f}")
    print("  ⭐ Sketch: kichik xotira, oldindan ma'lum xato, birlashtiriladi")


if __name__ == "__main__":
    main()

=== 1. HyperLogLog: kunlik noyob foydalanuvchilar ===
  hodisalar 1,500,000, aniq noyob 335,460
  aniq (Python set): 26,170,312 bayt
    p  registr    bayt       baho    xato  nazariya
    8      256     256    329,306  -1.83%     6.50%
   10    1,024   1,024    327,440  -2.39%     3.25%
   12    4,096   4,096    335,930  +0.14%     1.62%
   14   16,384  16,384    336,431  +0.29%     0.81%

=== 2. HLL xatosi: 30 xil xesh tuzi bo'yicha (p=10 va 12) ===
  p=10: o'rtacha xato +0.482%, RMS 3.309%, nazariya 3.250%
  p=12: o'rtacha xato -0.175%, RMS 1.580%, nazariya 1.625%
  ikki kun birlashmasi: aniq 546,072, HLL (max registr) 549,674, xato +0.66%

=== 3. Count-Min Sketch: mahsulot ko'rishlari ===
  ko'rishlar N=2,000,000, ko'rilgan mahsulotlar 82,570
      eps  d      w      bayt min xato  <=eps*N o_rt xato kamyob nisbiy
    0.001  1  2,719    10,876       25    0.964     728.9         444.6
    0.001  4  2,719    43,504       25    1.000     133.7          80.9
   0.0001  4 27,183   434,928        0    1.000       4.0           2.4
  top-10 mos kelishi (eps=0.0001, d=4): 10/10
  aniq lug'at: 82,570 kalit x (8+8) bayt ~ 1,321,120 bayt (kamida)

=== 4. Bloom filter: qora ro'yxatdagi telefonlar ===
  ro'yxatda n=100,000, tekshiriladigan begona 300,000
  bit/el   k      bayt  FPR o_lchangan  FPR nazariya  yolg_on manfiy
       4   3    50,000          0.1464        0.1469               0
       8   6   100,000          0.0220        0.0216               0
      10   7   125,000          0.0082        0.0082               0
      16  11   200,000          0.0005        0.0005               0
       8   1   100,000          0.1179        0.1175               0
       8  12   100,000          0.0487        0.0483               0
  optimal k = (m/n) ln2: 8 bit -> 5.5, 10 bit -> 6.9
  ⭐ Sketch: kichik xotira, oldindan ma'lum xato, birlashtiriladi

"""Map-reduce (ketma-ket simulyatsiya) va sqlite: indeks, EXPLAIN QUERY PLAN."""

import sqlite3
import tempfile
import zlib
from collections import Counter, defaultdict
from contextlib import closing
from pathlib import Path

import numpy as np
import pandas as pd

ASOSIY = ["va", "juda", "yaxshi", "mahsulot", "narx", "sifat", "tez",
          "yetkazib", "berildi", "yomon", "arzon", "qimmat", "kuryer",
          "buyurtma", "keldi", "kech", "rahmat", "tavsiya", "qilaman", "emas"]
BOGINLAR = ["ka", "lo", "mi", "ra", "su", "to", "na", "bi", "de", "yo"]


def hujjatlar_yarat(rng, n):
    """Sintetik mahsulot sharhlari (Zipf taqsimotli so'zlar)."""
    qolgan = sorted({a + b + c for a in BOGINLAR for b in BOGINLAR
                     for c in BOGINLAR})[:380]
    lugat = ASOSIY + qolgan
    vazn = 1 / np.arange(1, len(lugat) + 1) ** 1.0
    vazn /= vazn.sum()
    uzunlik = rng.integers(5, 21, n)
    sozlar = rng.choice(len(lugat), int(uzunlik.sum()), p=vazn)
    chegaralar = np.cumsum(uzunlik)[:-1]
    return [" ".join(lugat[i] for i in qism)
            for qism in np.split(sozlar, chegaralar)]


def bolim(kalit, r):
    """Deterministik bo'limlash (Python hash() jarayonlar orasida farq qiladi)."""
    return zlib.crc32(kalit.encode()) % r


def map_reduce(splitlar, mapper, r, kombinator=None, reducer=sum):
    stat = {"map": 0, "shuffle": 0}
    kirish = [defaultdict(list) for _ in range(r)]
    for split in splitlar:
        chiqish = [kv for yozuv in split for kv in mapper(yozuv)]
        stat["map"] += len(chiqish)
        if kombinator is not None:
            chiqish = kombinator(chiqish)
        for k, v in chiqish:                     # SHUFFLE
            kirish[bolim(k, r)][k].append(v)
            stat["shuffle"] += 1
    natija, yuk = {}, []
    for qism in kirish:                          # REDUCE
        yuk.append(sum(len(v) for v in qism.values()))
        for k in sorted(qism):
            natija[k] = reducer(qism[k])
    return natija, stat, yuk


def soz_mapper(hujjat):
    return [(s, 1) for s in hujjat.split()]


def yigindi_kombinator(juftlar):
    c = Counter()
    for k, v in juftlar:
        c[k] += v
    return sorted(c.items())


def qadamlar(db, sql, param=()):
    """So'rov bajarilishidagi VM qadamlar soni (100 qadam aniqlikda)."""
    hisob = [0]

    def f():
        hisob[0] += 1
        return 0

    db.set_progress_handler(f, 100)
    natija = db.execute(sql, param).fetchall()
    db.set_progress_handler(None, 0)
    return natija, hisob[0] * 100


def reja(db, sql, param=()):
    return " | ".join(r[3] for r in db.execute("EXPLAIN QUERY PLAN " + sql, param))


def hajm(db):
    return (db.execute("PRAGMA page_count").fetchone()[0]
            * db.execute("PRAGMA page_size").fetchone()[0])


def main() -> None:
    rng = np.random.default_rng(0)
    hujjatlar = hujjatlar_yarat(rng, 20_000)
    splitlar = [hujjatlar[i::8] for i in range(8)]

    print("=== 1. So'z sanash: map -> shuffle -> reduce (8 split, 4 reducer) ===")
    aniq = Counter(s for h in hujjatlar for s in h.split())
    for nom, komb in [("kombinatorsiz", None), ("kombinator bilan",
                                                 yigindi_kombinator)]:
        natija, stat, yuk = map_reduce(splitlar, soz_mapper, 4, komb)
        print(f"  {nom:<17} map {stat['map']:>7,}  shuffle {stat['shuffle']:>7,}"
              f"  reducer yuki {yuk}  aniq bilan mos: {natija == dict(aniq)}")
    eng = aniq.most_common(3)
    print(f"  eng ko'p so'zlar: {eng}, 'va' ulushi "
          f"{eng[0][1] / sum(aniq.values()):.3f}")

    print("\n=== 2. Reducer soni va issiq kalit (kombinatorsiz) ===")
    for r in [2, 4, 8, 16]:
        _, stat, yuk = map_reduce(splitlar, soz_mapper, r)
        print(f"  R={r:>2}: eng og'ir / o'rtacha yuk = "
              f"{max(yuk) / np.mean(yuk):.2f}, eng og'ir {max(yuk):,}")

    print("\n=== 3. Guruh o'rtachasi: noto'g'ri va to'g'ri kombinator ===")
    shahar = rng.choice(["Toshkent", "Samarqand", "Buxoro"], 60_000,
                        p=[0.6, 0.25, 0.15])
    summa = np.round(rng.lognormal(11, 0.8, 60_000), -2)
    summa[:5_000] *= 1.8                 # 1-split: bayram kuni, xaridlar katta
    yozuvlar = list(zip(shahar.tolist(), summa.tolist()))
    splitlar2 = [yozuvlar[:5_000], yozuvlar[5_000:45_000], yozuvlar[45_000:]]
    mapper = lambda y: [y]

    def ortacha_komb(juftlar):           # NOTO'G'RI: split o'rtachasi
        g = defaultdict(list)
        for k, v in juftlar:
            g[k].append(v)
        return [(k, float(np.mean(g[k]))) for k in sorted(g)]

    def juft_komb(juftlar):              # TO'G'RI: (yig'indi, sanoq)
        g = defaultdict(lambda: [0.0, 0])
        for k, v in juftlar:
            g[k][0] += v
            g[k][1] += 1
        return [(k, tuple(g[k])) for k in sorted(g)]

    yomon, _, _ = map_reduce(splitlar2, mapper, 2, ortacha_komb,
                             lambda vs: float(np.mean(vs)))
    yaxshi, _, _ = map_reduce(splitlar2, mapper, 2, juft_komb,
                              lambda vs: sum(v[0] for v in vs)
                              / sum(v[1] for v in vs))
    haqiqiy = pd.Series(summa).groupby(shahar).mean()
    for sh in haqiqiy.index:
        print(f"  {sh:<10} haqiqiy {haqiqiy[sh]:>9,.1f}  o'rtachalar "
              f"o'rtachasi {yomon[sh]:>9,.1f}  (sum, n) {yaxshi[sh]:>9,.1f}")

    print("\n=== 4. sqlite: indeks va EXPLAIN QUERY PLAN ===")
    n = 300_000
    dokon = rng.choice([f"D{i:02d}" for i in range(1, 21)], n)
    kun = rng.integers(1, 366, n)
    mijoz = rng.integers(1, 50_001, n)
    summa = np.round(rng.lognormal(11, 0.8, n), -2)
    qatorlar = list(zip(range(1, n + 1), dokon.tolist(), kun.tolist(),
                        mijoz.tolist(), summa.tolist()))
    q_mijoz = "SELECT COUNT(*), SUM(summa) FROM savdo WHERE mijoz_id = ?"
    q_oraliq = ("SELECT COUNT(*), SUM(summa) FROM savdo "
                "WHERE dokon = ? AND kun BETWEEN ? AND ?")
    q_keng = "SELECT COUNT(*), SUM(summa) FROM savdo WHERE kun >= ?"
    q_keng_scan = ("SELECT COUNT(*), SUM(summa) FROM savdo NOT INDEXED "
                   "WHERE kun >= ?")
    with tempfile.TemporaryDirectory() as papka:
        with closing(sqlite3.connect(Path(papka) / "savdo.db")) as db:
            db.execute("CREATE TABLE savdo (id INTEGER PRIMARY KEY, "
                       "dokon TEXT, kun INTEGER, mijoz_id INTEGER, summa REAL)")
            db.executemany("INSERT INTO savdo VALUES (?, ?, ?, ?, ?)", qatorlar)
            db.commit()
            h0 = hajm(db)
            r1, s1 = qadamlar(db, q_mijoz, (4242,))
            r2, s2 = qadamlar(db, q_oraliq, ("D07", 100, 130))
            print(f"  jadval: {n:,} qator, fayl {h0:,} bayt")
            print(f"  [indekssiz] mijoz: {reja(db, q_mijoz, (4242,))}; "
                  f"qadam {s1:,}")
            print(f"  [indekssiz] oraliq: {reja(db, q_oraliq, ('D07', 100, 130))}; "
                  f"qadam {s2:,}")
            db.execute("CREATE INDEX idx_mijoz ON savdo(mijoz_id)")
            db.execute("CREATE INDEX idx_dokon_kun ON savdo(dokon, kun)")
            db.execute("CREATE INDEX idx_kun ON savdo(kun)")
            db.commit()
            r1b, s1b = qadamlar(db, q_mijoz, (4242,))
            r2b, s2b = qadamlar(db, q_oraliq, ("D07", 100, 130))
            print(f"  [indeks]    mijoz: {reja(db, q_mijoz, (4242,))}; "
                  f"qadam {s1b:,}")
            print(f"  [indeks]    oraliq: "
                  f"{reja(db, q_oraliq, ('D07', 100, 130))}; qadam {s2b:,}")
            print(f"  natijalar bir xil: {r1 == r1b and r2 == r2b}; mijoz 4242: "
                  f"{r1b[0][0]} xarid; D07, 100-130 kun: {r2b[0][0]} chek")
            print(f"  fayl indekslar bilan: {hajm(db):,} bayt "
                  f"(+{hajm(db) / h0 - 1:.0%})")
            k1, s3 = qadamlar(db, q_keng, (2,))
            k2, s4 = qadamlar(db, q_keng_scan, (2,))
            print(f"  past selektivlik (kun >= 2, {k1[0][0] / n:.1%} qator): "
                  f"indeks {s3:,}, to'liq skan {s4:,} qadam")
            print(f"    reja: {reja(db, q_keng, (2,))}")
            sql_g = dict(db.execute("SELECT dokon, SUM(summa) FROM savdo "
                                    "GROUP BY dokon").fetchall())
    pd_g = pd.Series(summa).groupby(dokon).sum()
    farq = max(abs(sql_g[d] - pd_g[d]) / pd_g[d] for d in pd_g.index)
    print(f"  SQL GROUP BY va pandas groupby: maks nisbiy farq {farq:.1e}")
    print("  ⭐ Shuffle - eng qimmat; indeks - selektiv so'rov uchun")


if __name__ == "__main__":
    main()

=== 1. So'z sanash: map -> shuffle -> reduce (8 split, 4 reducer) ===
  kombinatorsiz     map 249,780  shuffle 249,780  reducer yuki [22889, 109676, 69343, 47872]  aniq bilan mos: True
  kombinator bilan  map 249,780  shuffle   3,200  reducer yuki [632, 800, 840, 928]  aniq bilan mos: True
  eng ko'p so'zlar: [('va', 38186), ('juda', 18768), ('yaxshi', 12490)], 'va' ulushi 0.153

=== 2. Reducer soni va issiq kalit (kombinatorsiz) ===
  R= 2: eng og'ir / o'rtacha yuk = 1.26, eng og'ir 157,548
  R= 4: eng og'ir / o'rtacha yuk = 1.76, eng og'ir 109,676
  R= 8: eng og'ir / o'rtacha yuk = 2.03, eng og'ir 63,411
  R=16: eng og'ir / o'rtacha yuk = 3.66, eng og'ir 57,162

=== 3. Guruh o'rtachasi: noto'g'ri va to'g'ri kombinator ===
  Buxoro     haqiqiy  88,038.1  o'rtachalar o'rtachasi 103,592.8  (sum, n)  88,038.1
  Samarqand  haqiqiy  88,420.5  o'rtachalar o'rtachasi 105,064.4  (sum, n)  88,420.5
  Toshkent   haqiqiy  87,765.2  o'rtachalar o'rtachasi 105,433.4  (sum, n)  87,765.2

=== 4. sqlite: indeks va EXPLAIN QUERY PLAN ===
  jadval: 300,000 qator, fayl 6,569,984 bayt
  [indekssiz] mijoz: SCAN savdo; qadam 900,000
  [indekssiz] oraliq: SCAN savdo; qadam 945,600
  [indeks]    mijoz: SEARCH savdo USING INDEX idx_mijoz (mijoz_id=?); qadam 100
  [indeks]    oraliq: SEARCH savdo USING INDEX idx_dokon_kun (dokon=? AND kun>? AND kun<?); qadam 9,300
  natijalar bir xil: True; mijoz 4242: 8 xarid; D07, 100-130 kun: 1320 chek
  fayl indekslar bilan: 17,534,976 bayt (+167%)
  past selektivlik (kun >= 2, 99.7% qator): indeks 1,795,100, to'liq skan 2,096,700 qadam
    reja: SEARCH savdo USING INDEX idx_kun (kun>?)
  SQL GROUP BY va pandas groupby: maks nisbiy farq 0.0e+00
  ⭐ Shuffle - eng qimmat; indeks - selektiv so'rov uchun

df.memory_usage().sum()                                         # ⚠️ matn ko'rinmaydi
df.memory_usage(deep=True, index=False).sum()                   # ✅

df["jami"] = df["miqdor"] * df["narx_ming"]                     # ⚠️ int8 * int16 -> to'lib ketadi
df["jami"] = df["miqdor"].astype("int64") * df["narx_ming"]     # ✅ avval kengaytiring

df = pd.read_csv("cheklar.csv")                                 # ⚠️ hamma ustun, standart turlar
df = pd.read_csv("cheklar.csv", usecols=kerakli, dtype=turlar)  # ✅

ortacha = np.mean([b["summa"].mean() for b in bolaklar])        # ⚠️
ortacha = sum(b["summa"].sum() for b in bolaklar) / sum(len(b) for b in bolaklar)  # ✅

var = (s2 - s * s / n) / (n - 1)                                # ⚠️ katastrofik ayirish
w = Welford(); [w.qosh(x) for x in oqim]; var = w.dispersiya    # ✅

reducer = hash(kalit) % R                                       # ⚠️ har jarayonda boshqacha
reducer = zlib.crc32(kalit.encode()) % R                        # ✅

for u in ustunlar: db.execute(f"CREATE INDEX i_{u} ON savdo({u})")    # ⚠️
db.execute("CREATE INDEX i_dk ON savdo(dokon, kun)")            # ✅ so'rovlarga qarab
# va EXPLAIN QUERY PLAN bilan tekshiring

1.  # 20 MB CSV standart pd.read_csv da xotirada taxminan qancha?
2.  # memory_usage() va memory_usage(deep=True) qaysi ustunlarda farq qiladi?
3.  # 400 000 ta noyob chek kodini category qilsak?
4.  # int8 ustun * 20 - qiymat 12 bo'lsa natija?
5.  # bo'lak medianalarining o'rtachasi = mediana?
6.  # qiymatlar ~1e9, std 50 - naiv dispersiya formulasi?
7.  # "har yangi element 1/2 ehtimol bilan" namunasi qaysi qismdan ko'p?
8.  # HLL p=12 - nisbiy xato taxminan?
9.  # Count-Min baho haqiqiydan kichik bo'lishi mumkinmi?
10. # Bloom filter "yo'q" desa?
11. # kombinator bilan shuffle hajmi qanday o'zgaradi?
12. # WHERE kun >= 2 (99.7% qator) - indeks qancha yutuq beradi?

1.  df = pd.read_csv("cheklar.csv"); print(df.memory_usage().sum())

2.  ortacha = np.mean([b.summa.mean() for b in pd.read_csv(yol, chunksize=10**5)])

3.  var = (sum(x * x for x in oqim) - sum(oqim) ** 2 / n) / (n - 1)

4.  reducer = hash(soz) % 8

5.  namuna = oqim[-1000:]   # "tasodifiy" namuna