Akış işleme ve mesaj kuyrukları: Kafka kavramları

Neden önemli? Gece 02:09'daki 80 bin liralık harcama

Ilgaz Bank'ın işlem tablosunda (bank_txns) gece yarısından sonra internetten yapılmış, 45 ile 94 bin lira arasında kart harcamaları var. Bunları ertesi sabah toplu (batch) raporda görmek, dolandırıcılık ekibi için "çok geç" demektir: kart çoktan boşalmıştır. Bu derste olaylar geldiği anda işlenir: işlem gerçekleşir, mesaj bir kuyruğa yazılır, saniyeler içinde bir tüketici onu okur, kuralı uygular, dakikalık sayacı günceller ve gerekirse alarm üretir. Bu dünyanın merkezinde Apache Kafka vardır; bankacılık, e-ticaret, lojistik ve oyun şirketlerinin çoğu olay akışlarını Kafka ya da benzeri bir sistemle taşır.

Kafka bir küme ve ağ ister, tarayıcıda çalışmaz. Gerçek kodu gri kutularda göreceksin; kavramları ise bellek içi bir log ve asyncio.Queue ile kendin kuracaksın: topic, partition, offset, tüketici grubu, teslim garantileri, pencereleme ve watermark. İş akışı orkestrasyonu: cron, DAG ve Airflow dersinde zamanlanmış toplu işleri, asyncio: eşzamansız kod, coroutine, await ve gather dersinde kuyruk ve işçi havuzunu gördün; burada ikisini birleştirip sürekli akan veriye geçiyoruz.

Toplu, mikro-toplu ve akış; olay zamanı ile işlem zamanı

Akışta iki saat vardır. Olay zamanı (event time) işlemin gerçekten olduğu andır (txn_time). İşlem zamanı (processing time) mesajın senin kodunla karşılaştığı andır. Mobil uygulama çevrimdışıyken yapılan bir ödeme 3 dakika sonra gelebilir; "dakikalık işlem sayısı" gibi iş ölçüleri olay zamanına göre hesaplanmalıdır, yoksa geciken olaylar yanlış dakikaya yazılır.

Kafka'nın temel parçaları

Topic (konu) bir olay türünün adıdır: bank.txns. Her topic birden çok partition'a bölünür; partition, yalnız sonuna ekleme yapılan, değişmez bir log'dur. Her mesajın partition içindeki sıra numarasına offset denir. Mesaj anahtarı (ör. hesap numarası) partition'ı belirler: aynı hesabın işlemleri hep aynı partition'a gider ve o partition içinde sırası korunur. Partition'lar arasında ise genel bir sıra yoktur.

import veri

txns = veri.kayitlar("ilgaz_bank", "bank_txns")


class Konu:
    """Kafka topic'inin bellek içi benzetimi: her partition bir liste (log)."""
    def __init__(self, ad, n_partition):
        self.ad = ad
        self.partitionlar = [[] for _ in range(n_partition)]

    def yaz(self, anahtar, deger):
        p = anahtar % len(self.partitionlar)
        self.partitionlar[p].append(deger)
        return p, len(self.partitionlar[p]) - 1        # (partition, offset)

    def oku(self, p, offset, maks=3):
        return list(enumerate(self.partitionlar[p][offset:offset + maks], start=offset))


konu = Konu("bank.txns", 3)
for t in txns[:12]:
    konu.yaz(t["account_id"], t)

print([len(p) for p in konu.partitionlar])
for offset, t in konu.oku(1, 0):
    print(offset, t["account_id"], t["txn_time"], t["amount"])

Saklama süresi (retention) mesajların ne kadar tutulacağını belirler (ör. 7 gün ya da 500 GB). Mesaj okundu diye silinmez; bu, bir kuyruktan (RabbitMQ) en büyük farkıdır. Yeni bir ekip aynı topic'i baştan okuyabilir, hata düzeltildikten sonra dünkü offset'e dönüp yeniden işleyebilirsin (replay).

Tüketici grubu (consumer group) aynı işi paylaşan tüketicilerdir. Kafka her partition'ı gruptaki tek tüketiciye atar; böylece bir partition'ın sırası korunur. 6 partition ve 3 tüketici varsa her biri 2 partition okur; 8 tüketici varsa 2'si boşta bekler. Paralelliğin üst sınırı partition sayısıdır. Grup, her partition için okuduğu son konumu commit eder (__consumer_offsets topic'ine yazılır). Tüketici çöküp yeniden başlarsa commit edilen offset'ten devam eder. Bir tüketici gruba girip çıktığında atamaların yeniden dağıtılmasına rebalance denir.

# pip install confluent-kafka      (Kafka'yı yerelde: docker run -p 9092:9092 apache/kafka:3.8.0)
from confluent_kafka import Producer, Consumer
import json

uretici = Producer({"bootstrap.servers": "localhost:9092", "enable.idempotence": True, "acks": "all"})
uretici.produce("bank.txns", key=str(t["account_id"]), value=json.dumps(t))
uretici.flush()

tuketici = Consumer({
    "bootstrap.servers": "localhost:9092",
    "group.id": "dolandiricilik-v1",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,          # commit'i biz yapacağız
})
tuketici.subscribe(["bank.txns"])
while True:
    msg = tuketici.poll(1.0)
    if msg is None or msg.error():
        continue
    isle(json.loads(msg.value()))          # önce işle
    tuketici.commit(msg, asynchronous=False)   # sonra commit: en az bir kez

Teslim garantileri ve idempotent tüketici

Pratikte en sık seçilen yol "en az bir kez + idempotent tüketici"dir: aynı mesaj iki kez gelse de sonuç değişmez. Bunun için her olayın benzersiz bir kimliği (txn_id) olur ve tüketici işlediği kimlikleri hatırlar ya da hedef tabloya INSERT ... ON CONFLICT DO NOTHING ile yazar.

# Ağ hatası yüzünden bazı mesajlar iki kez teslim edildi
teslimat = txns[:5] + txns[2:4] + txns[5:8]
islenen = set()
toplam = 0.0
for t in teslimat:
    if t["txn_id"] in islenen:        # idempotentlik: gördüysen atla
        continue
    islenen.add(t["txn_id"])
    toplam += t["amount"]
print(len(teslimat), "teslimat,", len(islenen), "benzersiz işlem, toplam:", round(toplam, 2))

asyncio.Queue ile üretici ve tüketiciler

Tek süreç içinde Kafka'nın üretici-tüketici ilişkisini asyncio.Queue ile kurabilirsin. Kuyruk dolarsa put bekler (geri basınç, backpressure); boşsa get bekler. Bitişi haber vermek için her tüketiciye bir "zehirli hap" (sentinel, ör. None) gönderilir. Bu derste olay döngüsünü new_event_loop().run_until_complete ile çalıştırıyoruz; betikte asyncio.run(main()) aynı işi yapar.

import asyncio


async def uretici(kuyruk, olaylar, n_tuketici):
    for o in olaylar:
        await kuyruk.put(o)
    for _ in range(n_tuketici):
        await kuyruk.put(None)                   # bitti sinyali


async def tuketici(ad, kuyruk, sayac):
    while (o := await kuyruk.get()) is not None:
        sayac[ad] = sayac.get(ad, 0) + 1
        await asyncio.sleep(0)                   # sırayı başkasına ver


async def main():
    kuyruk = asyncio.Queue(maxsize=10)
    sayac = {}
    await asyncio.gather(uretici(kuyruk, txns[:100], 3),
                         *(tuketici(f"t{i}", kuyruk, sayac) for i in range(3)))
    return sayac


loop = asyncio.new_event_loop()
sonuc = loop.run_until_complete(main())
loop.close()
print(sorted(sonuc.items()), sum(sonuc.values()))

Pencereleme: tumbling, sliding, session

Sonsuz bir akışta "toplam" anlamsızdır; bunun yerine zamanı pencerelere (window) bölersin.

Tumbling pencerede bir olayın penceresi, zamanın pencere boyutuna göre aşağı yuvarlanmasıyla bulunur. Dakikalık pencere için ISO tarih metninin ilk 16 karakteri yeter:

from collections import Counter

gun = [t for t in txns if t["txn_time"].startswith("2024-03-15")]
dakika = Counter(t["txn_time"][:16] for t in gun)
print(len(gun), "işlem,", len(dakika), "dolu dakika")
print("en yoğun dakikalar:", dakika.most_common(2))

Geç gelen veri ve watermark

Olaylar sırasız gelebilir. Dakikalık pencereyi ne zaman "kapandı" sayıp sonucu yayınlayacaksın? Sonsuza kadar beklersen sonuç hiç çıkmaz; hemen yayınlarsan geç gelenler kaybolur. Watermark bu ikilemin uzlaşmasıdır: "şimdiye kadar gördüğüm en büyük olay zamanı eksi izin verilen gecikme". Watermark bir pencerenin bitişini geçince pencere kapanır ve yayınlanır; o pencereye sonradan gelen olay geç sayılır (atılır, ayrı bir yan çıkışa yazılır ya da düzeltme olarak gönderilir).

olaylar = [("10:00:05", "a"), ("10:00:40", "b"), ("10:01:10", "c"),
           ("10:00:50", "d"), ("10:02:30", "e"), ("10:00:59", "f")]

def sn(s):
    h, m, x = map(int, s.split(":"))
    return h * 3600 + m * 60 + x

gecikme = 30
en_buyuk = 0
for zaman, ad in olaylar:
    en_buyuk = max(en_buyuk, sn(zaman))
    watermark = en_buyuk - gecikme
    pencere_sonu = sn(zaman) // 60 * 60 + 60
    durum = "GEÇ" if pencere_sonu <= watermark else "zamanında"
    print(ad, zaman, "watermark =", watermark - sn("10:00:00"), "sn ->", durum)

"d" olayı sırasız gelmesine rağmen 30 saniyelik tolerans içinde kaldı; "f" ise 10:02:30 görüldükten sonra 10:00 penceresine düştüğü için geç sayıldı. Gecikme toleransı büyüdükçe doğruluk artar ama sonuç gecikir: bu, akış işlemenin temel takasıdır.

Şema evrimi: Avro ve şema kayıt defteri

Üretici mesaja yeni bir alan eklediğinde eski tüketiciler kırılmamalıdır. Bu yüzden kurumsal Kafka kurulumlarında mesajlar JSON yerine şemalı bir biçimle (Avro, Protobuf) yazılır ve şemalar bir şema kayıt defterinde (Schema Registry) sürümlenir. Kayıt defteri yeni şemayı kaydetmeden önce uyumluluğu denetler: geriye uyumlu (backward) değişiklikte yeni şemayla eski veri okunabilir; bunun için eklenen her alanın varsayılan değeri olmalıdır.

{"type": "record", "name": "Txn", "namespace": "ilgazbank",
 "fields": [
   {"name": "txn_id", "type": "long"},
   {"name": "account_id", "type": "long"},
   {"name": "amount", "type": "double"},
   {"name": "channel", "type": "string"},
   {"name": "risk_skoru", "type": ["null", "double"], "default": null}   // v2: varsayılanlı yeni alan
 ]}

RabbitMQ, Redis Streams ve Kafka

CDC: veritabanından ambara akış

Change Data Capture (değişiklik yakalama), veritabanının işlem günlüğünü (PostgreSQL WAL, MySQL binlog) okuyup her INSERT, UPDATE ve DELETE'i bir olaya çevirir. Debezium bunu Kafka Connect üzerinde yapar; olaylar Kafka'ya, oradan ambara (Snowflake, BigQuery) ya da göle akar. Gece toplu kopyalama yerine dakikalar içinde güncel bir ambar elde edilir ve kaynak veritabanına sorgu yükü binmez.

{"op": "u", "ts_ms": 1718000000000, "source": {"table": "bank_accounts", "lsn": 2451},
 "before": {"account_id": 100019, "status": "aktif", "credit_limit": 20000},
 "after":  {"account_id": 100019, "status": "aktif", "credit_limit": 35000}}