Oqimli qayta ishlash (streaming)

diagramma(
    {
        "Hodisalar\n(to'xtovsiz)": (0, 1.6, 2.2, 1.1, "manba"),
        "Oyna\n(19:00-20:00)": (3.2, 2.9, 2.3, 0.9, "qayta"),
        "Oyna\n(20:00-21:00)": (3.2, 1.6, 2.3, 0.9, "qayta"),
        "Oyna\n(21:00-22:00)": (3.2, 0.3, 2.3, 0.9, "qayta"),
        "Watermark\n(chegarani belgilaydi)": (6.8, 1.6, 2.8, 1.1, "kuzatuv"),
        "Natija": (10.2, 1.6, 2.0, 1.1, "xizmat"),
    },
    [
        ("Hodisalar\n(to'xtovsiz)", "Oyna\n(19:00-20:00)"),
        ("Hodisalar\n(to'xtovsiz)", "Oyna\n(20:00-21:00)"),
        ("Hodisalar\n(to'xtovsiz)", "Oyna\n(21:00-22:00)"),
        ("Oyna\n(20:00-21:00)", "Watermark\n(chegarani belgilaydi)"),
        ("Watermark\n(chegarani belgilaydi)", "Natija"),
    ],
    nom="oqim_oyna.png", kenglik=14,
    izoh="Cheksiz oqimni oynalarga bo'lamiz. Watermark oynani qachon yopishni hal qiladi.",
)

import pandas as pd
show(pd.DataFrame([
    {"Vaqt turi": "Hodisa vaqti", "Manbai": "Hodisaning o'zida yozilgan",
     "Afzalligi": "Haqiqatni aks ettiradi", "Kamchiligi": "Kechikib kelishi mumkin"},
    {"Vaqt turi": "Qayta ishlash vaqti", "Manbai": "Serverning soati",
     "Afzalligi": "Hech qachon kechikmaydi", "Kamchiligi": "Haqiqatni buzadi"},
]))

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 "broker ${i} soniyada tayyor"; break; }
  sleep 1
done

import pandas as pd
show(pd.DataFrame([
    {"Oyna turi": "Tumbling (ketma-ket)", "Ta'rifi": "Ustma-ust tushmaydi",
     "Misol": "Har soat: 19-20, 20-21", "Qachon": "Soatlik hisobot"},
    {"Oyna turi": "Sliding (siljuvchi)", "Ta'rifi": "Ustma-ust tushadi",
     "Misol": "Oxirgi 1 soat, har 10 daqiqada", "Qachon": "Trend kuzatish"},
    {"Oyna turi": "Session (seans)", "Ta'rifi": "Jimlik bilan ajraladi",
     "Misol": "Foydalanuvchi 30 daq jim tursa", "Qachon": "Sayt tashrifi"},
]))

"""Oqimli qayta ishlash: oyna, watermark, kech kelgan ma'lumot."""
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Any


def oyna_boshi(vaqt: datetime, daqiqa: int) -> datetime:
    """Hodisa vaqtini oyna boshiga yaxlitlaydi."""
    jami = vaqt.hour * 60 + vaqt.minute
    boshi = (jami // daqiqa) * daqiqa
    return vaqt.replace(hour=boshi // 60, minute=boshi % 60, second=0, microsecond=0)


class OynaliHisob:
    """Oyna bo'yicha sanaydi va watermark bilan yopadi.

    kechikish: watermark hodisa vaqtidan qancha orqada yuradi.
    Katta qiymat — kam yo'qotish, ko'p kutish. Kichik — teskarisi.
    """

    def __init__(self, oyna_daqiqa: int = 60, kechikish_daqiqa: int = 10) -> None:
        self.oyna_daqiqa = oyna_daqiqa
        self.kechikish = timedelta(minutes=kechikish_daqiqa)
        self.oynalar: dict[datetime, int] = defaultdict(int)
        self.yopilgan: dict[datetime, int] = {}
        self.watermark: datetime | None = None
        self.tashlangan = 0

    def qabul(self, vaqt: datetime) -> None:
        """Bitta hodisani qabul qiladi."""
        if self.watermark is None or vaqt - self.kechikish > self.watermark:
            self.watermark = vaqt - self.kechikish
            self._yop()
        oyna = oyna_boshi(vaqt, self.oyna_daqiqa)
        if oyna in self.yopilgan:
            self.tashlangan += 1          # kech keldi, oyna allaqachon yopilgan
            return
        self.oynalar[oyna] += 1

    def _yop(self) -> None:
        """Watermark dan oldingi oynalarni yopadi."""
        chegara = self.watermark - timedelta(minutes=self.oyna_daqiqa)
        for oyna in [o for o in self.oynalar if o < chegara]:
            self.yopilgan[oyna] = self.oynalar.pop(oyna)

    def hammasi(self) -> dict[str, Any]:
        return {
            "yopilgan_oynalar": len(self.yopilgan),
            "ochiq_oynalar": len(self.oynalar),
            "tashlangan_hodisa": self.tashlangan,
        }

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

a = KafkaAdminClient(bootstrap_servers='localhost:19092')
if 'hodisa' not in a.list_topics():
    a.create_topics([NewTopic(name='hodisa', num_partitions=1, replication_factor=1)])

d = pd.read_csv('../datasets/ballon.csv').head(5000)
p = KafkaProducer(bootstrap_servers='localhost:19092',
                  value_serializer=lambda v: json.dumps(v).encode())
for r in d.itertuples():
    p.send('hodisa', value={'vaqt': r.vaqt, 'ombor': r.ombor})
p.flush()
print(f'{len(d):,} hodisa yuborildi')
" 2>&1 | grep -v Warning

python3 -c "
import sys, json; sys.path.insert(0, 'src')
from datetime import datetime
from kafka import KafkaConsumer
from quvur.oqim import OynaliHisob

h = OynaliHisob(oyna_daqiqa=60, kechikish_daqiqa=10)
c = KafkaConsumer('hodisa', bootstrap_servers='localhost:19092',
                  auto_offset_reset='earliest', group_id='oyna1',
                  consumer_timeout_ms=6000)
n = 0
for m in c:
    v = json.loads(m.value)
    h.qabul(datetime.fromisoformat(v['vaqt']))
    n += 1
c.close()

print(f'qayta ishlangan hodisa: {n:,}')
for k, val in h.hammasi().items():
    print(f'  {k:22s}: {val:,}')
" 2>&1 | grep -v Warning

python3 -c "
import sys, json; sys.path.insert(0, 'src')
from datetime import datetime
from kafka import KafkaConsumer
from quvur.oqim import OynaliHisob

hodisalar = []
c = KafkaConsumer('hodisa', bootstrap_servers='localhost:19092',
                  auto_offset_reset='earliest', group_id='oyna2',
                  consumer_timeout_ms=6000)
for m in c:
    hodisalar.append(datetime.fromisoformat(json.loads(m.value)['vaqt']))
c.close()

# hodisalar tartibsiz kelsin (haqiqiy oqimdagidek)
import random
random.seed(42)
aralash = hodisalar[:]
for i in range(0, len(aralash) - 50, 50):
    j = i + random.randint(1, 49)
    aralash[i], aralash[j] = aralash[j], aralash[i]

sarlavha = ['kechikish', 'yopilgan', 'tashlangan', 'yoqotish']
print(f'{sarlavha[0]:>12s} {sarlavha[1]:>10s} {sarlavha[2]:>12s} {sarlavha[3]:>10s}')
print('-' * 48)
for k in (0, 1, 5, 10, 30, 60):
    h = OynaliHisob(oyna_daqiqa=60, kechikish_daqiqa=k)
    for v in aralash:
        h.qabul(v)
    r = h.hammasi()
    ulush = r['tashlangan_hodisa'] / len(aralash) * 100
    print(f'{k:>10d}daq {r[\"yopilgan_oynalar\"]:>10,d} {r[\"tashlangan_hodisa\"]:>12,d} {ulush:>9.2f}%')
" 2>&1 | grep -v Warning

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