CDC: ma'lumotlar bazasi o'zgarishlarini ushlash

SELECT * FROM elon WHERE yangilandi > '<oxirgi tekshiruv>';

diagramma(
    {
        "Postgres\n(RostUy)": (0, 1.6, 2.2, 1.1, "manba"),
        "WAL\n(tranzaksiya jurnali)": (3.0, 1.6, 2.4, 1.1, "qayta"),
        "Replikatsiya\nsloti": (6.2, 1.6, 2.2, 1.1, "qayta"),
        "Kafka\n(10-bob)": (9.2, 1.6, 2.0, 1.1, "saqlash"),
        "Polling\n(SELECT ...)": (3.0, -0.2, 2.4, 1.0, "xizmat"),
    },
    [
        ("Postgres\n(RostUy)", "WAL\n(tranzaksiya jurnali)", "har commit"),
        ("WAL\n(tranzaksiya jurnali)", "Replikatsiya\nsloti"),
        ("Replikatsiya\nsloti", "Kafka\n(10-bob)"),
        ("Postgres\n(RostUy)", "Polling\n(SELECT ...)", "har N daqiqa"),
    ],
    nom="cdc_yol.png", kenglik=13,
    izoh="Ikki yo'l: pastdagisi oson va kamchilikli, yuqoridagisi WAL ni to'g'ridan o'qiydi.",
)

docker rm -f kitob-pg > /dev/null 2>&1
docker run -d --name kitob-pg -p 15455:5432 \
  -e POSTGRES_PASSWORD=kitob -e POSTGRES_DB=rostuy \
  postgres:16 -c wal_level=logical -c max_replication_slots=4 -c max_wal_senders=4 > /dev/null
echo "konteyner ishga tushdi"

for i in $(seq 1 60); do
  docker exec kitob-pg pg_isready -U postgres > /dev/null 2>&1 && { echo "baza ${i} soniyada tayyor"; break; }
  sleep 1
done
docker exec kitob-pg psql -U postgres -d rostuy -tAc "SHOW wal_level;" | sed 's/^/wal_level: /'

python3 -c "
import psycopg2, pandas as pd

d = pd.read_csv('../datasets/ijara.csv').drop_duplicates('elon_id').head(2000)
c = psycopg2.connect(host='localhost', port=15455, user='postgres', password='kitob', dbname='rostuy')
c.autocommit = True
k = c.cursor()
k.execute('''
DROP TABLE IF EXISTS elon;
CREATE TABLE elon (
  elon_id     text PRIMARY KEY,
  tuman       text,
  narx_mln    numeric,
  xonalar     int,
  holat       text DEFAULT 'faol',
  yangilandi  timestamp DEFAULT now()
);''')
qatorlar = [(r.elon_id, r.tuman, float(r.narx_mln), int(r.xonalar)) for r in d.itertuples()]
k.executemany('INSERT INTO elon (elon_id,tuman,narx_mln,xonalar) VALUES (%s,%s,%s,%s)', qatorlar)
k.execute('SELECT count(*) FROM elon')
print('jadvaldagi e\\'lonlar:', k.fetchone()[0])
c.close()
"

python3 -c "
import psycopg2
c = psycopg2.connect(host='localhost', port=15455, user='postgres', password='kitob', dbname='rostuy')
c.autocommit = True
k = c.cursor()
k.execute(\"SELECT count(*) FROM pg_replication_slots WHERE slot_name='kitob_slot'\")
if k.fetchone()[0] == 0:
    k.execute(\"SELECT pg_create_logical_replication_slot('kitob_slot','test_decoding')\")
    print('slot yaratildi:', k.fetchone()[0])
else:
    print('slot allaqachon bor')
k.execute('SELECT slot_name, plugin, active FROM pg_replication_slots')
print('slotlar:', k.fetchall())
c.close()
"

python3 -c "
import psycopg2, collections
c = psycopg2.connect(host='localhost', port=15455, user='postgres', password='kitob', dbname='rostuy')
c.autocommit = True
k = c.cursor()

k.execute('SELECT now()')
oxirgi_tekshiruv = k.fetchone()[0]

# --- ikki tekshiruv orasidagi kun ---
k.execute('''INSERT INTO elon (elon_id,tuman,narx_mln,xonalar)
             SELECT 'Y'||g, 'Yunusobod', 5.0, 2 FROM generate_series(1,50) g''')
k.execute(\"UPDATE elon SET narx_mln=narx_mln*1.1, yangilandi=now() WHERE elon_id IN (SELECT elon_id FROM elon ORDER BY elon_id LIMIT 30)\")
k.execute(\"UPDATE elon SET narx_mln=narx_mln*1.2, yangilandi=now() WHERE elon_id IN (SELECT elon_id FROM elon ORDER BY elon_id LIMIT 30)\")
k.execute(\"DELETE FROM elon WHERE elon_id IN (SELECT elon_id FROM elon ORDER BY elon_id DESC LIMIT 20)\")

haqiqiy = 50 + 30 + 30 + 20
print(f'haqiqatda sodir bo\\'lgan o\\'zgarishlar: {haqiqiy}')
print('  50 INSERT, 30 UPDATE, yana 30 UPDATE (bir xil qatorlar), 20 DELETE')
print()

# --- 1-usul: polling ---
k.execute('SELECT count(*) FROM elon WHERE yangilandi > %s', (oxirgi_tekshiruv,))
polling = k.fetchone()[0]
print(f'POLLING ko\\'rdi : {polling} qator')

# --- 2-usul: CDC ---
k.execute(\"SELECT data FROM pg_logical_slot_get_changes('kitob_slot',NULL,NULL)\")
hodisalar = [d for (d,) in k.fetchall()]
tur = collections.Counter()
for h in hodisalar:
    for t in ('INSERT','UPDATE','DELETE'):
        if f': {t}:' in h: tur[t] += 1
print(f'CDC     ko\\'rdi : {sum(tur.values())} o\\'zgarish -> {dict(tur)}')
print()
print(f'polling o\\'tkazib yuborgan: {haqiqiy - polling} ta o\\'zgarish')
c.close()
"

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

"""Postgres WAL -> Kafka. Minimal, lekin haqiqiy CDC quvuri."""
import json
import re
from typing import Any, Iterator

QATOR = re.compile(r"table (\S+): (INSERT|UPDATE|DELETE): (.*)")
MAYDON = re.compile(r"(\w+)\[[\w ]+\]:('(?:[^']|'')*'|\S+)")


def tahlil(qator: str) -> dict[str, Any] | None:
    """test_decoding matnini dict ga aylantiradi. BEGIN/COMMIT -> None."""
    m = QATOR.match(qator)
    if not m:
        return None
    jadval, amal, qolgan = m.groups()
    qiymat = {}
    for nom, xom in MAYDON.findall(qolgan):
        if xom.startswith("'"):
            qiymat[nom] = xom[1:-1].replace("''", "'")
        elif xom == "null":
            qiymat[nom] = None
        else:
            qiymat[nom] = xom
    return {"jadval": jadval, "amal": amal, "qiymat": qiymat}


def oqim(kursor, slot: str, limit: int | None = None) -> Iterator[dict[str, Any]]:
    """Slotdagi o'zgarishlarni o'qiydi va tahlil qilib beradi."""
    kursor.execute(
        "SELECT data FROM pg_logical_slot_get_changes(%s, NULL, %s)", (slot, limit)
    )
    for (xom,) in kursor.fetchall():
        h = tahlil(xom)
        if h:
            yield h


def kafka_xabari(hodisa: dict[str, Any]) -> tuple[str, bytes]:
    """Kalit = birlamchi kalit, shuning uchun bir e'lon bitta partitionda qoladi."""
    kalit = hodisa["qiymat"].get("elon_id", "")
    return kalit, json.dumps(hodisa, ensure_ascii=False).encode()

python3 -c "
import sys, collections; sys.path.insert(0, 'src')
import psycopg2
from kafka import KafkaProducer
from quvur.cdc import oqim, kafka_xabari

c = psycopg2.connect(host='localhost', port=15455, user='postgres', password='kitob', dbname='rostuy')
c.autocommit = True
k = c.cursor()

# yangi o'zgarishlar yaratamiz
k.execute(\"UPDATE elon SET holat='sotildi' WHERE tuman='Chilonzor'\")
k.execute(\"DELETE FROM elon WHERE elon_id LIKE 'Y%'\")

p = KafkaProducer(bootstrap_servers='localhost:19092')
tur = collections.Counter()
for h in oqim(k, 'kitob_slot'):
    kalit, qiymat = kafka_xabari(h)
    p.send('elon-cdc', key=kalit.encode(), value=qiymat)
    tur[h['amal']] += 1
p.flush()
print('Kafka ga yuborildi:', dict(tur), '-> jami', sum(tur.values()))
c.close()
" 2>&1 | grep -v Warning

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

c = KafkaConsumer('elon-cdc', bootstrap_servers='localhost:19092',
                  auto_offset_reset='earliest', group_id='ombor', consumer_timeout_ms=5000)
tur = collections.Counter()
namuna = {}
for m in c:
    h = json.loads(m.value)
    tur[h['amal']] += 1
    namuna.setdefault(h['amal'], h)
c.close()

print('Kafka dan o\\'qildi:', dict(tur))
print()
for amal in ('UPDATE', 'DELETE'):
    if amal in namuna:
        q = namuna[amal]['qiymat']
        print(f'{amal:7s} namuna: {dict(list(q.items())[:4])}')
" 2>&1 | grep -v Warning

import pandas as pd
show(pd.DataFrame([
    {"Bosqich": "1. Snapshot", "Nima qiladi": "Jadvalni to'liq o'qib ko'chiradi",
     "Qachon": "Bir marta, boshida", "Manba": "SELECT *"},
    {"Bosqich": "2. Oqim", "Nima qiladi": "Faqat o'zgarishlarni uzatadi",
     "Qachon": "Doimiy", "Manba": "WAL sloti"},
]))

python3 -c "
import psycopg2
c = psycopg2.connect(host='localhost', port=15455, user='postgres', password='kitob', dbname='rostuy')
c.autocommit = True
c.cursor().execute(\"SELECT pg_drop_replication_slot('kitob_slot')\")
c.close()
print('slot o\\'chirildi')
" 2>/dev/null || echo "slot allaqachon yo'q"
docker rm -f kitob-pg kitob-kafka > /dev/null 2>&1 && echo "konteynerlar o'chirildi"
docker ps --format '{{.Names}}'