Apache Iceberg: jadval formati

ombor/elon/
  2026-09-01.parquet
  2026-09-02.parquet
  ...

diagramma(
    {
        "Katalog\n(jadval -> metadata)": (0, 1.6, 2.6, 1.1, "qayta"),
        "Metadata\n(sxema, snapshotlar)": (3.4, 1.6, 2.6, 1.1, "qayta"),
        "Manifest\n(qaysi fayllar)": (6.8, 1.6, 2.4, 1.1, "saqlash"),
        "data-1.parquet": (10.0, 3.0, 2.2, 0.9, "manba"),
        "data-2.parquet": (10.0, 1.7, 2.2, 0.9, "manba"),
        "data-3.parquet": (10.0, 0.4, 2.2, 0.9, "manba"),
    },
    [
        ("Katalog\n(jadval -> metadata)", "Metadata\n(sxema, snapshotlar)"),
        ("Metadata\n(sxema, snapshotlar)", "Manifest\n(qaysi fayllar)"),
        ("Manifest\n(qaysi fayllar)", "data-1.parquet"),
        ("Manifest\n(qaysi fayllar)", "data-2.parquet"),
        ("Manifest\n(qaysi fayllar)", "data-3.parquet"),
    ],
    nom="iceberg_qatlam.png", kenglik=14,
    izoh="Ma'lumot baribir Parquet. Iceberg ustiga uch qatlam metama'lumot qo'yadi.",
)

"""Iceberg katalogi bilan ishlash. SQLite katalogi — mahalliy, lekin haqiqiy."""
from pathlib import Path

from pyiceberg.catalog.sql import SqlCatalog

OMBOR = Path("ombor")


def katalog(nom: str = "kitob") -> SqlCatalog:
    """Mahalliy katalog ochadi. Papkalar yo'q bo'lsa yaratiladi."""
    OMBOR.mkdir(parents=True, exist_ok=True)
    return SqlCatalog(
        nom,
        uri=f"sqlite:///{OMBOR.absolute()}/katalog.db",
        warehouse=f"file://{OMBOR.absolute()}",
    )


def snapshotlar(jadval) -> list[tuple[int, int]]:
    """(snapshot_id, qatorlar soni) ro'yxati — eskisidan yangisiga."""
    return [
        (s.snapshot_id, int(s.summary.get("total-records", 0)))
        for s in jadval.metadata.snapshots
    ]

python3 -c "
import sys, shutil, pathlib; sys.path.insert(0, 'src')
shutil.rmtree('ombor', ignore_errors=True)
import pandas as pd, pyarrow as pa
from quvur.muz import katalog

USTUN = ['elon_id','tuman','narx_mln','xonalar','tavsif_uzunligi','broker']
d = pd.read_csv('../datasets/ijara.csv').drop_duplicates('elon_id')
print(f'manba: {len(d):,} e\\'lon')

c = katalog()
c.create_namespace_if_not_exists('rostuy')
birinchi = pa.Table.from_pandas(d.head(5000)[USTUN], preserve_index=False)
t = c.create_table('rostuy.elon', schema=birinchi.schema)
t.append(birinchi)
print(f'jadval yaratildi, qatorlar: {len(t.scan().to_arrow()):,}')
print(f'sxema: {[f.name for f in t.schema().fields]}')
"

python3 -c "
import sys; sys.path.insert(0, 'src')
import pandas as pd, pyarrow as pa
from quvur.muz import katalog, snapshotlar

USTUN = ['elon_id','tuman','narx_mln','xonalar','tavsif_uzunligi','broker']
d = pd.read_csv('../datasets/ijara.csv').drop_duplicates('elon_id')
c = katalog()
t = c.load_table('rostuy.elon')

t.append(pa.Table.from_pandas(d.iloc[5000:8000][USTUN], preserve_index=False))
t.append(pa.Table.from_pandas(d.iloc[8000:9500][USTUN], preserve_index=False))

print('SNAPSHOTLAR:')
for i, (sid, n) in enumerate(snapshotlar(t), 1):
    print(f'  {i}. id={sid}  qatorlar={n:,}')
print()
print(f'hozirgi holat: {len(t.scan().to_arrow()):,} qator')
"

python3 -c "
import sys; sys.path.insert(0, 'src')
from quvur.muz import katalog, snapshotlar

c = katalog()
t = c.load_table('rostuy.elon')
snaps = snapshotlar(t)

print('O\\'TMISHGA SAYOHAT:')
for i, (sid, _) in enumerate(snaps, 1):
    ark = t.scan(snapshot_id=sid).to_arrow()
    d = ark.to_pandas()
    print(f'  {i}-snapshot: {len(ark):>6,} qator, broker ulushi {d[\"broker\"].mean():.3f}')
"

python3 -c "
import sys; sys.path.insert(0, 'src')
import pandas as pd, pyarrow as pa
from pyiceberg.types import LongType
from quvur.muz import katalog

c = katalog()
t = c.load_table('rostuy.elon')
print('ESKI sxema:', [f.name for f in t.schema().fields])

with t.update_schema() as u:
    u.add_column('rasmlar_soni', LongType(), 'E\\'londagi rasmlar soni')
t = c.load_table('rostuy.elon')
print('YANGI sxema:', [f.name for f in t.schema().fields])
print()

eski = t.scan().to_arrow().to_pandas()
print(f'eski qatorlar: {len(eski):,}')
print(f'yangi ustun eski qatorlarda: {eski[\"rasmlar_soni\"].isna().sum():,} ta null')
"

python3 -c "
import sys; sys.path.insert(0, 'src')
import pandas as pd, pyarrow as pa
from quvur.muz import katalog

USTUN = ['elon_id','tuman','narx_mln','xonalar','tavsif_uzunligi','broker','rasmlar_soni']
d = pd.read_csv('../datasets/ijara.csv').drop_duplicates('elon_id')
c = katalog()
t = c.load_table('rostuy.elon')

yangi = d.iloc[9500:11000][USTUN]
t.append(pa.Table.from_pandas(yangi, preserve_index=False).cast(t.schema().as_arrow()))

ark = t.scan().to_arrow().to_pandas()
print(f'jami qatorlar        : {len(ark):,}')
print(f'rasmlar_soni bor     : {ark[\"rasmlar_soni\"].notna().sum():,}')
print(f'rasmlar_soni null    : {ark[\"rasmlar_soni\"].isna().sum():,}')
print()
print('Bitta jadvalda ikki avlod ma\\'lumot birga yashayapti.')
"

python3 -c "
import sys, pathlib, time; sys.path.insert(0, 'src')
import pandas as pd, pyarrow as pa
from quvur.muz import katalog

USTUN = ['elon_id','tuman','narx_mln','xonalar','tavsif_uzunligi','broker']
d = pd.read_csv('../datasets/ijara.csv').drop_duplicates('elon_id').head(6000)
c = katalog()
c.create_namespace_if_not_exists('sinov')
try: c.drop_table('sinov.kichik')
except Exception: pass
try: c.drop_table('sinov.katta')
except Exception: pass

# 60 ta kichik yozuv (oqim uslubi)
ar = pa.Table.from_pandas(d[USTUN], preserve_index=False)
kichik = c.create_table('sinov.kichik', schema=ar.schema)
for i in range(0, 6000, 100):
    kichik.append(pa.Table.from_pandas(d.iloc[i:i+100][USTUN], preserve_index=False))

# bitta katta yozuv (batch uslubi)
katta = c.create_table('sinov.katta', schema=ar.schema)
katta.append(ar)

def sana(t, nom):
    fayllar = [f.file.file_path for f in t.scan().plan_files()]
    hajm = sum(pathlib.Path(f.replace('file://','')).stat().st_size for f in fayllar)
    t0 = time.perf_counter()
    for _ in range(5): t.scan().to_arrow()
    ms = (time.perf_counter()-t0)/5*1000
    print(f'{nom:22s} fayllar={len(fayllar):>4d}  jami={hajm/1024:>8.1f} KB  o\\'qish={ms:>7.1f} ms')
    return ms, len(fayllar), hajm

a, fa, ha = sana(kichik, '60 ta kichik yozuv')
b, fb, hb = sana(katta, '1 ta katta yozuv')
print()
print(f'fayl soni  {fa/fb:.0f}x ko\\'p')
print(f'disk hajmi {ha/hb:.1f}x katta')
print(f'o\\'qish     {a/b:.1f}x sekin')
"

import pandas as pd
show(pd.DataFrame([
    {"Muammo": "Qaysi fayllar jadvalga kiradi", "Parquet to'plami": "Noma'lum",
     "Iceberg": "Manifest aytadi"},
    {"Muammo": "Yozayotganda o'qish", "Parquet to'plami": "Yarim ma'lumot ko'rinadi",
     "Iceberg": "Snapshot atomik almashadi"},
    {"Muammo": "Kecha qanday edi", "Parquet to'plami": "Javob yo'q",
     "Iceberg": "Time travel"},
    {"Muammo": "Ustun qo'shish", "Parquet to'plami": "Hamma faylni qayta yozish",
     "Iceberg": "Faqat metama'lumot"},
    {"Muammo": "Bitta qatorni o'chirish", "Parquet to'plami": "Faylni qayta yozish",
     "Iceberg": "Delete fayl"},
]))

python3 -c "
import sys; sys.path.insert(0, 'src')
from quvur.muz import katalog
c = katalog()
for t in ('sinov.kichik', 'sinov.katta'):
    try: c.drop_table(t); print(f'{t} o\\'chirildi')
    except Exception as e: print(f'{t}: {type(e).__name__}')
print('rostuy.elon saqlanadi (16-bobda kerak)')
"