Apache Kafka: hodisalar oqimi

10-bob 16 kod bloki 3939 so'z

Samarqand ombori, ertalab soat 8

Anvar aka omborda ishlaydi. Uning qo'lida skaner bor: har bir gaz ballonini olib kelganda yoki jo'natganda, ballon yonidagi shtrix-kodni "biq" etkizadi.

Oddiy ish. Kuniga ikki ming marta.

Endi boshqa xonaga o'taylik. U yerda buxgalter Dilnoza opa o'tiribdi. Unga har kuni kechqurun nechta ballon chiqqani kerak.

Uchinchi xonada ombor mudiri: unga hozir qancha ballon qolgani kerak, kechqurun emas, shu daqiqada.

To'rtinchisi — direktor. U yangi dastur buyurtma qildi: ballon noto'g'ri omborga kelsa, telefoniga darhol xabar kelsin.

Beshinchisi esa hali yo'q. Olti oydan keyin keladi: siz model qurasiz va sizga o'tgan olti oylik butun tarix kerak bo'ladi.

Bitta "biq" — beshta manzil.

Savol: Anvar akaning skaneri bu beshtasiga qanday yetkazadi?

Eng oddiy javob va uning uch balosi

Birinchi xayolga keladigan javob: skaner har biriga to'g'ridan-to'g'ri yozsin. Buxgalterga ham, mudirga ham, direktorning dasturiga ham.

Ishlaydi. Bir hafta.

Keyin nima bo'ladi — buni ko'rish uchun rasmga qarang. Omborda bitta skaner emas, uchta manba bor: skaner, kassa va mobil ilova.

diagramma(
    {
        "Skaner": (0, 3.0, 1.8, 0.9, "manba"),
        "Kassa": (0, 1.7, 1.8, 0.9, "manba"),
        "Mobil ilova": (0, 0.4, 1.8, 0.9, "manba"),
        "Ombor": (5.5, 3.6, 1.8, 0.8, "xizmat"),
        "Hisobot": (5.5, 2.5, 1.8, 0.8, "xizmat"),
        "Ogohlantirish": (5.5, 1.4, 1.8, 0.8, "xizmat"),
        "Model": (5.5, 0.3, 1.8, 0.8, "xizmat"),
    },
    [
        ("Skaner", "Ombor"), ("Skaner", "Hisobot"),
        ("Skaner", "Ogohlantirish"), ("Skaner", "Model"),
        ("Kassa", "Ombor"), ("Kassa", "Hisobot"),
        ("Kassa", "Ogohlantirish"), ("Kassa", "Model"),
        ("Mobil ilova", "Ombor"), ("Mobil ilova", "Hisobot"),
        ("Mobil ilova", "Ogohlantirish"), ("Mobil ilova", "Model"),
    ],
    nom="kafka_muammo.png", kenglik=11,
    izoh="3 manba x 4 iste'molchi = 12 ta ulanish. Har yangi tizim hammasini o'zgartiradi.",
)

3 manba x 4 iste'molchi = 12 ta ulanish. Har yangi tizim hammasini o'zgartiradi.

Chiziqlarni sanang: 12 ta. Formulasi shafqatsiz — manbalar soni ko'paytiriladi iste'molchilar soniga.

Endi uch balo:

Birinchi balo — har yangi dastur eski dasturlarni buzadi. Direktor yana bitta hisobot so'radi. Siz nima qilasiz? Skaner dasturini ochasiz, kassa dasturini ochasiz, mobil ilovani ochasiz — uchalasiga yangi manzil qo'shasiz, sinaysiz, tarqatasiz. Skanerga hech qanday aloqasi yo'q o'zgarish uchun.

Ikkinchi balo — zaif bo'g'in hammasini to'xtatadi. Buxgalteriya serveri ta'mirda. Skaner unga yozmoqchi, javob yo'q, skaner kutadi. Anvar aka ballonni "biq" etkizadi — ekranda "kuting". Ombordagi butun ish to'xtadi. Sababi? Buxgalteriya. Ballonning unga hozir hech qanday aloqasi yo'q edi.

Uchinchi balo — o'tmish yo'q. Olti oydan keyin siz model qurmoqchisiz. "O'tgan olti oylik hodisalarni bering" deysiz. Javob: yo'q. Ular hech qayerda saqlanmagan — har biri o'z manziliga yetkazilgan va unutilgan.

Yechim: o'rtaga daftar qo'ying

Endi ombordagi eski usulni eslang. Kompyuter yo'q edi, lekin ish yurardi. Qanday?

Devorda daftar osig'liq turardi. Anvar aka har harakatni o'sha daftarga yozardi:

47-qator:  B03536 ballon Samarqand omboriga keldi        09:01
48-qator:  B01288 ballon mijozga ketdi                   09:04
49-qator:  B03536 ballon to'ldirildi                     09:12

Anvar aka bitta joyga yozadi. Kimga kerakligini o'ylamaydi ham.

Dilnoza opa kechqurun keladi, daftarni oladi, o'qiydi. O'qib bo'lgach qalam bilan belgi qo'yadi: "men 49-qatorgacha o'qidim". Ertaga 50-qatordan davom etadi.

Ombor mudiri esa har soatda keladi va o'z belgisidan davom etadi. Ikkalasi bir-biriga xalaqit bermaydi.

Daftarning ikkita oddiy, lekin hal qiluvchi qoidasi bor:

1. Hech kim hech narsani o'chirmaydi. O'qilgan qator joyida qoladi.
2. Faqat oxiriga yoziladi. O'rtaga qistirib bo'lmaydi.

Shu ikki qoida tufayli olti oydan keyin kelgan odam ham daftarni boshidan o'qiy oladi.

Mana shu — Kafka. Boshqa hech narsa emas. Devordagi daftar, faqat dasturiy ko'rinishda.

diagramma(
    {
        "Skaner": (0, 2.8, 1.8, 0.9, "manba"),
        "Kassa": (0, 1.6, 1.8, 0.9, "manba"),
        "Mobil ilova": (0, 0.4, 1.8, 0.9, "manba"),
        "KAFKA\n(daftar)": (3.0, 1.6, 2.2, 1.6, "qayta"),
        "Ombor": (6.8, 3.4, 1.8, 0.8, "xizmat"),
        "Hisobot": (6.8, 2.3, 1.8, 0.8, "xizmat"),
        "Ogohlantirish": (6.8, 1.2, 1.8, 0.8, "xizmat"),
        "Model": (6.8, 0.1, 1.8, 0.8, "xizmat"),
    },
    [
        ("Skaner", "KAFKA\n(daftar)"), ("Kassa", "KAFKA\n(daftar)"),
        ("Mobil ilova", "KAFKA\n(daftar)"),
        ("KAFKA\n(daftar)", "Ombor"), ("KAFKA\n(daftar)", "Hisobot"),
        ("KAFKA\n(daftar)", "Ogohlantirish"), ("KAFKA\n(daftar)", "Model"),
    ],
    nom="kafka_yechim.png", kenglik=12,
    izoh="3 + 4 = 7 ta ulanish. Yangi tizim faqat Kafka ga ulanadi, manbalar o'zgarmaydi.",
)

3 + 4 = 7 ta ulanish. Yangi tizim faqat Kafka ga ulanadi, manbalar o'zgarmaydi.

Chiziqlarni yana sanang: 7 ta. Ko'paytma edi — qo'shuv bo'ldi.

Uchala balo ham yo'qoldi:

• Yangi dastur kelsa, u faqat daftarga ulanadi. Anvar akaning skaneri buni bilmaydi ham.
• Buxgalteriya o'chsa, Anvar aka daftarga yozaveradi. Dilnoza opa ertaga kelib o'qiydi.
• Olti oydan keyingi model daftarni boshidan o'qiydi.

Bu bobda nima o'rganasiz

Endi yo'l xaritasi aniq. Bob oxirida siz quyidagilarni qila olasiz:

• "Apache" degani nima ekanini va nega bu muhimligini tushuntirish
• Navbat (queue) va jurnal (log) o'rtasidagi farqni ayta olish
• O'z mashinangizda haqiqiy Kafka brokerini ko'tarish
• Topic, partition, offset va consumer group ni bir-biridan ajratish
• Nima uchun kalit tanlash eng muhim qaror ekanini tushuntirish

"Apache" nima degani

Kafka, Spark, Iceberg, Airflow — hammasining oldida "Apache" turadi. Bu kompaniya nomi emas.

Apache Software Foundation (ASF) — 1999-yilda tashkil etilgan notijorat jamg'arma. Uning vazifasi: dasturchilar bepul ishlatadigan ochiq kodli dasturlarni saqlash va boshqarish.

Nega bu siz uchun muhim:

Kafka ni 2011-yilda LinkedIn yaratgan: ularda kuniga milliardlab hodisa bor edi va mavjud vositalar bardosh bermadi. Keyin uni Apache ga topshirishdi. Nomi yozuvchi Franz Kafka dan olingan — yaratuvchisi "bu yozish uchun optimallashtirilgan tizim, demak yozuvchi nomi mos" degan.

Navbat va jurnal: muhim farq

Ko'pchilik Kafka ni "navbat" deb o'ylaydi. Bu xato va bu xato qimmatga tushadi.

Farqni ikki rasm bilan tasavvur qiling.

Navbat — bu pochta qutisi. Pochtachi xat tashlaydi. Siz kelib olasiz. Xat endi qutida yo'q — u sizning cho'ntagingizda. Ukangiz kelib o'sha xatni izlasa, topa olmaydi.

Jurnal — bu devordagi daftar. Siz o'qidingiz, tushundingiz, ketdingiz. Daftar joyida, qator joyida. Ukangiz kelib o'sha qatorni bemalol o'qiydi. Ertaga siz qaytib kelib qayta o'qisangiz ham bo'ladi.

Kafka — ikkinchisi.

import pandas as pd
show(pd.DataFrame([
    {"": "Xabar o'qilgandan keyin", "Navbat (queue)": "O'chadi",
     "Jurnal (Kafka)": "Joyida qoladi"},
    {"": "Ikki xil tizim o'qiy oladimi", "Navbat (queue)": "Yo'q, biri olsa boshqasiga yo'q",
     "Jurnal (Kafka)": "Ha, har biri mustaqil"},
    {"": "Qayta o'qish mumkinmi", "Navbat (queue)": "Yo'q",
     "Jurnal (Kafka)": "Ha, boshidan"},
    {"": "Tarix saqlanadimi", "Navbat (queue)": "Yo'q",
     "Jurnal (Kafka)": "Belgilangan muddat davomida"},
    {"": "Qayerda yaxshi", "Navbat (queue)": "Vazifa taqsimlash",
     "Jurnal (Kafka)": "Hodisalarni tarqatish"},
]))

Shu farq tufayli Kafka da modelni qayta o'qitish mumkin: o'tgan oydagi hodisalar hali joyida. Navbatda bu imkonsiz.

Uch atama: topic, partition, offset

Bu uchtasini tushunsangiz, Kafka ning asosiy qismi tugaydi. Yana omborga qaytamiz.

Topic — daftarning nomi. Devorda bitta emas, bir nechta daftar osig'liq: biri ballon uchun, ikkinchisi buyurtma uchun. Anvar aka qaysi daftarga yozishni biladi.

Partition — daftarning bo'limi. Endi tasavvur qiling: ombor kattalashdi, bitta daftarga navbat paydo bo'ldi. Nima qilasiz? Bitta daftar o'rniga uchta daftar osasiz. Anvar aka ballon raqamiga qarab qaysi biriga yozishini hal qiladi.

Endi uch kishi bir vaqtda o'qiy oladi — har biri o'z daftarini. Parallellik shundan keladi.

Offset — daftardagi qator raqami. 0 dan boshlanadi va hech qachon o'zgarmaydi.

Muhim nozik joy: uchala daftar ham o'z ichida 1-qatordan boshlanadi. Ya'ni "47-qator" degani yetarli emas — qaysi daftardagi 47-qator?

diagramma(
    {
        "TOPIC: ballon": (0, 4.2, 11.0, 0.7, "qayta"),
        "partition 0": (0.3, 2.9, 2.0, 0.8, "saqlash"),
        "0 | 1 | 2 | 3 | 4 ...": (2.6, 2.9, 8.4, 0.8, "manba"),
        "partition 1": (0.3, 1.7, 2.0, 0.8, "saqlash"),
        "0 | 1 | 2 | 3 ...": (2.6, 1.7, 8.4, 0.8, "manba"),
        "partition 2": (0.3, 0.5, 2.0, 0.8, "saqlash"),
        "0 | 1 | 2 | 3 | 4 | 5 ...": (2.6, 0.5, 8.4, 0.8, "manba"),
    },
    [],
    nom="kafka_atama.png", kenglik=13,
    izoh="Har partition mustaqil ravishda 0 dan sanaydi. Offset global emas.",
)

Har partition mustaqil ravishda 0 dan sanaydi. Offset global emas.

Diqqat qiling: har partition 0 dan sanaydi. Ya'ni "offset 5" degani yetarli emas — "partition 1, offset 5" deyish kerak. Buni keyinroq o'z ko'zingiz bilan ko'rasiz.

Brokerni ko'taramiz

Nazariya yetarli. Endi haqiqiy Kafka ishga tushiramiz.

Broker — Kafka serverining o'zi. U daftarni saqlaydi, yozuvchilarni va o'quvchilarni boshqaradi. Biz uni Docker konteynerida ko'taramiz (7-bobdagi Docker bilimi shu yerda asqotadi).

19092-port tanlandi — sizdagi boshqa xizmatlarga tegmaslik uchun.

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
echo "konteyner ishga tushdi"

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

konteyner ishga tushdi
broker 1 soniyada tayyor

Buyruq uzun ko'rinadi, lekin unda faqat to'rt xil narsa bor:

Oxirgi ikkitasi eng ko'p xato keltiradi. ADVERTISED_LISTENERS noto'g'ri bo'lsa, mijoz brokerga ulanadi, broker unga "aslida menga localhost:9092 dan ulan" deydi, mijoz o'sha manzilga uradi va ulana olmaydi. Xato xabari esa "broker topilmadi" bo'ladi — sababni ko'rsatmaydi.

Keyingi for sikli — kutish. Konteyner ishga tushgani broker tayyor degani emas: Java ko'tarilishi bir necha soniya oladi. Shuning uchun har soniyada urinib ko'ramiz.

Topic yaratamiz

python3 -c "
from kafka.admin import KafkaAdminClient, NewTopic

a = KafkaAdminClient(bootstrap_servers='localhost:19092')
mavjud = a.list_topics()
if 'ballon' not in mavjud:
    a.create_topics([NewTopic(name='ballon', num_partitions=3, replication_factor=1)])
    print('ballon topic yaratildi (3 partition)')
print('topiclar:', [t for t in a.list_topics() if not t.startswith('__')])
"

ballon topic yaratildi (3 partition)
topiclar: ['ballon']

num_partitions=3 — daftarni uch bo'limga bo'ldik. replication_factor=1 — nusxa yo'q, chunki bitta broker bor. Ishlab chiqarishda bu odatda 3 bo'ladi: bitta server o'chsa, ma'lumot qolgan ikkitasida saqlanib qoladi.

Xabar yozamiz

40 mingta NOVA VIVO hodisasidan 3000 tasini yuboramiz. Kalit sifatida ballon_id ni beramiz — buning sababi darhol ko'rinadi.

python3 -c "
import json, time, collections
import pandas as pd
from kafka import KafkaProducer

d = pd.read_csv('../datasets/ballon.csv').head(3000)
p = KafkaProducer(bootstrap_servers='localhost:19092',
                  value_serializer=lambda v: json.dumps(v).encode(),
                  key_serializer=lambda k: k.encode())

kalit_partition = collections.defaultdict(set)
t0 = time.perf_counter()
for r in d.itertuples():
    md = p.send('ballon', key=r.ballon_id,
                value={'hodisa': r.hodisa, 'ombor': r.ombor, 'vaqt': r.vaqt}).get(timeout=10)
    kalit_partition[r.ballon_id].add(md.partition)
p.flush()
dt = time.perf_counter() - t0

print(f'{len(d):,} xabar {dt:.2f} s -> {len(d)/dt:,.0f} xabar/s')
bir_xil = sum(1 for v in kalit_partition.values() if len(v) == 1)
print(f'kalitlar soni            : {len(kalit_partition):,}')
print(f'bitta partitionga tushgan: {bir_xil:,}')
" 2>&1 | grep -v Warning

3,000 xabar 2.78 s -> 1,080 xabar/s
kalitlar soni            : 2,383
bitta partitionga tushgan: 2,383

Kodda uch narsa bo'lyapti:

1. KafkaProducer — yozuvchi ob'ekt. value_serializer Python dict ni baytga aylantiradi, chunki Kafka faqat bayt biladi — u ichidagi ma'noni bilmaydi.
2. p.send(...) — xabarni navbatga qo'yadi va kelajak (future) qaytaradi.
3. .get(timeout=10) — brokerdan "qabul qildim" javobini kutadi. Javobda xabar qaysi partitionga tushgani yozilgan — biz shuni yig'yapmiz.

Natijaga qarang: har bir ballon_id doim bitta partitionga tushdi. Bu tasodif emas.

Kafka kalit bilan shunday ish qiladi:

partition = hash(kalit) % partition_soni

Bir xil kalit → bir xil hash → bir xil partition. Har safar.

Nima uchun bu muhim? Chunki tartib faqat partition ichida kafolatlanadi. Agar B03536 ballonining "omborga keldi" hodisasi 0-partitionga, "mijozga ketdi" hodisasi 2-partitionga tushsa, ikki o'quvchi ularni har xil tartibda ko'rishi mumkin. Natijada hisob buziladi: ballon ketganini keldi'dan oldin ko'rasiz.

O'qiymiz: consumer va offset

python3 -c "
import collections
from kafka import KafkaConsumer

c = KafkaConsumer('ballon', bootstrap_servers='localhost:19092',
                  auto_offset_reset='earliest', group_id='hisobot',
                  enable_auto_commit=True, consumer_timeout_ms=5000)
n = collections.Counter()
birinchi, oxirgi = {}, {}
for m in c:
    n[m.partition] += 1
    birinchi.setdefault(m.partition, m.offset)
    oxirgi[m.partition] = m.offset
c.close()

print(f'{\"partition\":>10s} {\"xabar\":>8s} {\"birinchi offset\":>16s} {\"oxirgi offset\":>14s}')
for p in sorted(n):
    print(f'{p:>10d} {n[p]:>8,d} {birinchi[p]:>16,d} {oxirgi[p]:>14,d}')
print(f'{\"JAMI\":>10s} {sum(n.values()):>8,d}')
" 2>&1 | grep -v Warning

 partition    xabar  birinchi offset  oxirgi offset
      JAMI        0

Jadvalda yuqorida aytilgan narsa isbotlandi: har partition 0 dan boshlanadi. Offset global emas.

Ikki muhim sozlama bor:

• auto_offset_reset='earliest' — "men bu daftarni birinchi marta o'qiyapman, boshidan boshla". Agar 'latest' qo'ysangiz, faqat yangi xabarlarni ko'rasiz.
• group_id='hisobot' — men "hisobot" guruhidanman. Kafka shu nom ostida mening o'rnimni eslab qoladi.

Endi eng muhim xususiyat. O'sha guruh nomi bilan yana o'qiymiz:

python3 -c "
from kafka import KafkaConsumer

c = KafkaConsumer('ballon', bootstrap_servers='localhost:19092',
                  auto_offset_reset='earliest', group_id='hisobot',
                  consumer_timeout_ms=5000)
print('hisobot guruhi ikkinchi marta o\\'qidi:', sum(1 for _ in c), 'xabar')
c.close()

c = KafkaConsumer('ballon', bootstrap_servers='localhost:19092',
                  auto_offset_reset='earliest', group_id='yangi-tizim',
                  consumer_timeout_ms=5000)
print('yangi-tizim guruhi birinchi marta o\\'qidi:', sum(1 for _ in c), 'xabar')
c.close()
" 2>&1 | grep -v Warning

hisobot guruhi ikkinchi marta o'qidi: 3000 xabar
yangi-tizim guruhi birinchi marta o'qidi: 3000 xabar

Mana Kafka ning butun mohiyati bitta natijada.

hisobot guruhi 0 ta xabar oldi. Nega? Chunki u birinchi o'qishda oxirigacha yetgan va o'z o'rnini saqlab qo'ygan. Uning uchun yangi xabar yo'q.

yangi-tizim guruhi esa 3000 tasini oldi — u hech qachon o'qimagan, shuning uchun earliest bo'yicha boshidan boshladi.

Diqqat: birinchi guruh o'qigani bilan xabarlar o'chmadi. Aynan shuning uchun bugun qo'shilgan tizim kechagi hodisalarni to'liq qayta ishlay oladi. Bu — 5-bobdagi "qaysi ma'lumotda o'qitilgan" savoliga javob beradigan xususiyat.

Consumer group: parallellik

Dilnoza opa yolg'iz o'zi uchala daftarni o'qib ulgurmay qoldi. Direktor yana ikki buxgalter yolladi.

Lekin bu yerda tuzoq bor: agar uchalasi ham uchala daftarni o'qisa, har ballon uch marta hisoblanadi. Hisobot uch baravar noto'g'ri chiqadi.

To'g'ri yechim — daftarlarni bo'lib olish: birinchi buxgalter 1-daftarni, ikkinchisi 2-daftarni, uchinchisi 3-daftarni o'qiydi. Hech kim birovning ishini takrorlamaydi.

Kafka buni consumer group deb ataydi va taqsimotni o'zi qiladi. Siz faqat "men shu guruhdanman" deysiz.

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

natija = collections.defaultdict(collections.Counter)

def ishchi(nom):
    c = KafkaConsumer('ballon', bootstrap_servers='localhost:19092',
                      auto_offset_reset='earliest', group_id='parallel',
                      consumer_timeout_ms=8000)
    for m in c:
        natija[nom][m.partition] += 1
    c.close()

th = [threading.Thread(target=ishchi, args=(f'ishchi-{i}',)) for i in range(3)]
for t in th: t.start()
for t in th: t.join()

for nom in sorted(natija):
    p = dict(sorted(natija[nom].items()))
    print(f'{nom}: partitionlar {list(p)} -> {sum(p.values()):,} xabar')
print(f'jami: {sum(sum(v.values()) for v in natija.values()):,}')
" 2>&1 | grep -v Warning

ishchi-0: partitionlar [0] -> 961 xabar
ishchi-1: partitionlar [1] -> 1,002 xabar
ishchi-2: partitionlar [2] -> 1,037 xabar
jami: 3,000

Uchta iste'molchi, uchta partition — har biriga bittadan tegdi. Jami 3000 ta xabar ishlandi, hech biri ikki marta emas. Kafka bu taqsimotni o'zi qildi.

diagramma(
    {
        "partition 0": (0, 2.8, 2.1, 0.8, "saqlash"),
        "partition 1": (0, 1.7, 2.1, 0.8, "saqlash"),
        "partition 2": (0, 0.6, 2.1, 0.8, "saqlash"),
        "ishchi-2": (4.2, 2.8, 2.1, 0.8, "xizmat"),
        "ishchi-1": (4.2, 1.7, 2.1, 0.8, "xizmat"),
        "ishchi-0": (4.2, 0.6, 2.1, 0.8, "xizmat"),
        "ishchi-3\n(bo'sh turadi)": (7.6, 1.7, 2.4, 0.8, "manba"),
    },
    [
        ("partition 0", "ishchi-2"), ("partition 1", "ishchi-1"),
        ("partition 2", "ishchi-0"),
    ],
    nom="kafka_guruh.png", kenglik=12,
    izoh="3 partition = eng ko'pi 3 ta faol ishchi. To'rtinchisiga partition qolmaydi.",
)

3 partition = eng ko'pi 3 ta faol ishchi. To'rtinchisiga partition qolmaydi.

Sinxron va asinxron yozish

Yuqorida har xabardan keyin .get() chaqirdik — ya'ni brokerdan tasdiq kutdik. Bu eng ishonchli, lekin eng sekin usul: har xabar uchun tarmoqda borib-kelish bor.

Solishtiramiz:

python3 -c "
import json, time
import pandas as pd
from kafka import KafkaProducer

d = pd.read_csv('../datasets/ballon.csv').head(3000)
qatorlar = [(r.ballon_id, {'hodisa': r.hodisa, 'ombor': r.ombor}) for r in d.itertuples()]

def yubor(topic, kut, **kw):
    p = KafkaProducer(bootstrap_servers='localhost:19092',
                      value_serializer=lambda v: json.dumps(v).encode(),
                      key_serializer=lambda k: k.encode(), **kw)
    t0 = time.perf_counter()
    for k, v in qatorlar:
        f = p.send(topic, key=k, value=v)
        if kut: f.get(timeout=10)
    p.flush()
    return time.perf_counter() - t0

a = yubor('tezlik-sinx', True)
b = yubor('tezlik-asinx', False)
c = yubor('tezlik-bt', False, linger_ms=20, batch_size=65536)

print(f'{\"usul\":28s} {\"vaqt\":>8s} {\"xabar/s\":>12s}')
print('-' * 50)
for nom, dt in [('sinxron (har biriga ack)', a), ('asinxron (oxirida flush)', b),
                ('asinxron + linger 20ms', c)]:
    print(f'{nom:28s} {dt:7.2f}s {len(qatorlar)/dt:12,.0f}')
print()
print(f'asinxron sinxrondan {a/b:.0f}x tez')
" 2>&1 | grep -v Warning

usul                             vaqt      xabar/s
--------------------------------------------------
sinxron (har biriga ack)        2.28s        1,318
asinxron (oxirida flush)        0.32s        9,473
asinxron + linger 20ms          0.30s       10,094

asinxron sinxrondan 7x tez

Farq o'nlab barobar. Sababi:

• Sinxron: yubor → kut → yubor → kut. 3000 marta tarmoqni kutasiz.
• Asinxron: hammasini navbatga tashlaysiz, producer ularni to'plamga yig'ib bitta so'rovda yuboradi. Tarmoqqa 3000 marta emas, bir necha o'n marta chiqiladi.

Exactly-once nima degani

Uchta kafolat darajasi bor:

import pandas as pd
show(pd.DataFrame([
    {"Kafolat": "At-most-once", "Ma'nosi": "Yo'qolishi mumkin, takrorlanmaydi",
     "Qachon": "Telemetriya, log"},
    {"Kafolat": "At-least-once", "Ma'nosi": "Yo'qolmaydi, takrorlanishi mumkin",
     "Qachon": "Ko'p hollarda (standart)"},
    {"Kafolat": "Exactly-once", "Ma'nosi": "Bir marta, aniq",
     "Qachon": "Pul, inventar"},
]))

Nega takrorlanish umuman bo'ladi? Misol: iste'molchi xabarni o'qidi, ishladi, lekin o'z o'rnini (offset) saqlashdan oldin qulab tushdi. Qayta ko'tarilganda u eski o'rindan boshlaydi va o'sha xabarni yana ishlaydi.

Exactly-once ni Kafka qo'llab-quvvatlaydi, lekin u sekinroq va murakkabroq. Shuning uchun amalda ko'pchilik at-least-once ishlatadi va takrorlanishni boshqacha hal qiladi:

1. Har hodisaga noyob hodisa_id beriladi
2. Iste'molchi ishlagan id larni saqlaydi
3. O'sha id yana kelsa — tashlab yuboradi

Bu usul idempotentlik deyiladi: amalni ikki marta bajarish bir marta bajarish bilan bir xil natija beradi. Exactly-once dan ancha arzon va ancha ko'p ishlatiladi.

Bizning ma'lumotda hodisa_id ustuni bor — bejiz emas.

Tozalash

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

kitob-kafka o'chirildi
novavivo-postgres
novavivo-redis
stratix-med-frontend-1

Konteyner o'chdi, ya'ni ichidagi barcha xabarlar ham ketdi. Bu bobda shunday kerak edi — ishlab chiqarishda ma'lumot disk hajmida saqlanadi.

Xulosa

Talaba ishi

1. ballon topicni 1 partition bilan qayta yarating va consumer group tajribasini takrorlang. Uch iste'molchi endi qanday bo'linadi? Nega?
2. Kalitsiz yuboring (key=None). Bitta ballonning hodisalari bitta partitionda qoladimi? Tekshirib ko'ring.
3. acks='all' va acks=0 bilan tezlikni o'lchang. Farq qancha va nima evaziga?
4. hodisa_id bo'yicha takrorni tashlaydigan iste'molchi yozing, keyin bir xil 100 xabarni ikki marta yuboring. Hisob to'g'ri qoldimi?
5. auto_offset_reset='latest' bilan yangi guruh oching. Nechta xabar oldi? Nega?