Real vaqtda ML: oqimga model qo'shish

import pandas as pd
show(pd.DataFrame([
    {"": "Kim boshlaydi", "So'rov-javob": "Foydalanuvchi", "Oqimli": "Hodisa"},
    {"": "Kim kutadi", "So'rov-javob": "Foydalanuvchi", "Oqimli": "Hech kim"},
    {"": "Asosiy o'lchov", "So'rov-javob": "p99 kechikish", "Oqimli": "Orqada qolish (lag)"},
    {"": "Yuk oshsa", "So'rov-javob": "429 qaytaring (30-bob)", "Oqimli": "Navbat o'sadi, kutadi"},
    {"": "Misol", "So'rov-javob": "RostUy filtri", "Oqimli": "NOVA VIVO ballon hodisasi"},
]))

docker rm -f kitob-kafka > /dev/null 2>&1
docker run -d --name kitob-kafka -p 19092:9092 \
  -e KAFKA_NODE_ID=1 -e KAFKA_PROCESS_ROLES=broker,controller \
  -e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:19092 \
  -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
  -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
  -e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \
  -e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \
  -e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \
  apache/kafka:3.9.0 > /dev/null
for i in $(seq 1 60); do
  python3 -c "from kafka import KafkaAdminClient; KafkaAdminClient(bootstrap_servers='localhost:19092')" > /dev/null 2>&1 \
    && { echo "kafka ${i} soniyada tayyor"; break; }
  sleep 1
done
python3 -c "
import sys, pathlib; sys.path.insert(0, 'src')
if not pathlib.Path('model.onnx').exists():
    import pandas as pd
    from sklearn.ensemble import RandomForestClassifier
    from sklearn.model_selection import train_test_split
    from quvur.xususiyat import XUSUSIYATLAR, tayyorla
    from quvur.format import onnx_ga
    d = pd.read_csv('../datasets/ijara.csv')
    x, y = tayyorla(d), d['broker']
    Xtr, _, ytr, _ = train_test_split(x, y, test_size=0.25, stratify=y, random_state=42)
    m = RandomForestClassifier(n_estimators=150, min_samples_leaf=3,
                               random_state=42, n_jobs=1).fit(Xtr, ytr)
    onnx_ga(m, len(XUSUSIYATLAR), 'model.onnx')
print('model tayyor')
"

"""Oqimli bashorat: Kafka -> model -> Kafka.

3-bobdagi qoida bu yerda ham amal qiladi: xususiyat kodi O'SHA `tayyorla()`.
Oqimli tizimda alohida kod yozish - skewning eng keng tarqalgan sababi.
"""
import json
import time
from typing import Any

import pandas as pd

from quvur.format import onnx_ehtimol, onnx_sessiya
from quvur.xususiyat import XOM_USTUNLAR, tayyorla


def toplamni_bahola(sessiya, xabarlar: list[dict[str, Any]],
                    chegara: float) -> list[dict[str, Any]]:
    """To'plamni birgalikda baholaydi (29-bob: to'plam arzonroq)."""
    d = pd.DataFrame([x["elon"] for x in xabarlar], columns=XOM_USTUNLAR)
    e = onnx_ehtimol(sessiya, tayyorla(d).values)
    natija = []
    for xabar, ehtimol in zip(xabarlar, e):
        natija.append({
            "elon_id": xabar["elon_id"],
            "ehtimol": round(float(ehtimol), 4),
            "broker": bool(ehtimol >= chegara),
            "model_versiya": "v1",
            "vaqt": time.time(),
        })
    return natija

python3 -c "
import json
import pandas as pd
from kafka import KafkaProducer
from kafka.admin import KafkaAdminClient, NewTopic

a = KafkaAdminClient(bootstrap_servers='localhost:19092')
mavjud = a.list_topics()
for t in ('elon-kirish', 'elon-natija'):
    if t not in mavjud:
        a.create_topics([NewTopic(name=t, num_partitions=3, replication_factor=1)])

XUS = ['xonalar','maydon','qavat','jami_qavat','narx_mln','tavsif_uzunligi',
       'rasmlar_soni','telefon_takrorlanishi','kunlik_elonlar','egasi_tasdiqlangan']
d = pd.read_csv('../datasets/ijara.csv').drop_duplicates('elon_id').head(2000)

p = KafkaProducer(bootstrap_servers='localhost:19092',
                  value_serializer=lambda v: json.dumps(v).encode(),
                  key_serializer=lambda k: k.encode(),
                  linger_ms=20, batch_size=65536)
for r in d.itertuples():
    p.send('elon-kirish', key=r.elon_id,
           value={'elon_id': r.elon_id, 'elon': {u: getattr(r, u) for u in XUS}})
p.flush()
print(f'{len(d):,} elon oqimga yuborildi')
" 2>&1 | grep -v Warning

python3 -c "
import sys, json, time; sys.path.insert(0, 'src')
from kafka import KafkaConsumer, KafkaProducer
from quvur.format import onnx_sessiya
from xizmat.oqimli import toplamni_bahola

s = onnx_sessiya('model.onnx')
c = KafkaConsumer('elon-kirish', bootstrap_servers='localhost:19092',
                  auto_offset_reset='earliest', group_id='bashoratchi',
                  enable_auto_commit=False, consumer_timeout_ms=8000,
                  max_poll_records=200)
p = KafkaProducer(bootstrap_servers='localhost:19092',
                  value_serializer=lambda v: json.dumps(v).encode(),
                  key_serializer=lambda k: k.encode(),
                  linger_ms=20, batch_size=65536)

jami = broker = 0
toplam = []
t0 = time.perf_counter()

def jonat(toplam):
    global jami, broker
    if not toplam: return
    for n in toplamni_bahola(s, toplam, 0.5):
        p.send('elon-natija', key=n['elon_id'], value=n)
        broker += n['broker']
    jami += len(toplam)

for m in c:
    toplam.append(json.loads(m.value))
    if len(toplam) >= 200:
        jonat(toplam); toplam = []
jonat(toplam)
p.flush()
c.commit()
c.close()

dt = time.perf_counter() - t0
print(f'qayta ishlandi : {jami:,} elon')
print(f'broker dedi    : {broker:,} ({broker/max(jami,1)*100:.1f}%)')
print(f'vaqt           : {dt:.2f} s -> {jami/dt:,.0f} elon/s')
" 2>&1 | grep -v Warning

python3 -c "
import json, collections
from kafka import KafkaConsumer

c = KafkaConsumer('elon-natija', bootstrap_servers='localhost:19092',
                  auto_offset_reset='earliest', group_id='hisobot',
                  consumer_timeout_ms=8000)
n = collections.Counter()
namuna = None
for m in c:
    v = json.loads(m.value)
    n['broker' if v['broker'] else 'ega'] += 1
    if namuna is None and v['broker']: namuna = v
c.close()

print('natija oqimida:')
for k, v in sorted(n.items()):
    print(f'  {k:8s}: {v:>6,}')
print()
print('namuna xabar:', json.dumps(namuna, ensure_ascii=False))
" 2>&1 | grep -v Warning

python3 -c "
from kafka import KafkaConsumer, TopicPartition
c = KafkaConsumer(bootstrap_servers='localhost:19092', group_id='bashoratchi')
tps = [TopicPartition('elon-kirish', p) for p in c.partitions_for_topic('elon-kirish')]
c.assign(tps)
oxiri = c.end_offsets(tps)
jami_lag = 0
print(f'{\"partition\":>10s} {\"oxirgi\":>8s} {\"oqilgan\":>9s} {\"lag\":>6s}')
for tp in sorted(tps, key=lambda t: t.partition):
    komit = c.committed(tp) or 0
    lag = oxiri[tp] - komit
    jami_lag += lag
    print(f'{tp.partition:>10d} {oxiri[tp]:>8d} {komit:>9d} {lag:>6d}')
c.close()
print(f'{\"JAMI LAG\":>10s} {jami_lag:>25d}')
" 2>&1 | grep -v Warning

import pandas as pd
show(pd.DataFrame([
    {"Yechim": "Hodisaga qo'shib yuborish", "Kim hisoblaydi": "Manba tizimi",
     "Afzalligi": "Model sodda", "Kamchiligi": "Manba o'zgarishi kerak"},
    {"Yechim": "Oqimda hisoblash (14-bob)", "Kim hisoblaydi": "Oyna + holat",
     "Afzalligi": "Tez, yangi", "Kamchiligi": "Murakkab, holat saqlanadi"},
    {"Yechim": "Xususiyat ombori", "Kim hisoblaydi": "Alohida tizim",
     "Afzalligi": "O'qitish va xizmat bitta manba", "Kamchiligi": "Yana bitta tizim"},
]))

docker rm -f kitob-kafka > /dev/null 2>&1 && echo "kafka o'chirildi"
docker ps --format '{{.Names}}' | grep -v k3d