Kafka / Redpanda: olay günlüğü + Avro — DE Zoomcamp (Gençler)

Distributed Commit Log

Kafka, dağıtılmış, bölümlenmiş, tekrarlanabilir (replicated) bir commit log’dur. Bu tanımın her kelimesi önemli:

Commit log: sadece eklenen (append-only), asla değiştirilmeyen kayıt.
Bölümlenmiş (partitioned): bir topic, birden çok partition’a bölünür; her partition bağımsız bir log.
Dağıtılmış: partition’lar farklı broker’lara dağılır.
Tekrarlanabilir (replicated): her partition birden çok kopyasıyla yazılır (replication factor, genelde 3).

Topic: "siparisler"  (3 partition, replication factor 2)

Broker 1                          Broker 2                          Broker 3
┌──────────────────────┐          ┌──────────────────────┐          ┌──────────────────────┐
│ Partition 0 (leader) │◄────────►│ Partition 0 (follower)│          │                      │
│ #0 sipariş A         │          │ #0 sipariş A         │          │                      │
│ #1 sipariş B         │          │ #1 sipariş B         │          │                      │
│ #2 sipariş C         │          │ #2 sipariş C         │          │                      │
└──────────────────────┘          └──────────────────────┘          └──────────────────────┘
                                   ┌──────────────────────┐          ┌──────────────────────┐
                                   │ Partition 1 (follower)│◄────────►│ Partition 1 (leader) │
                                   │ #0 ödeme X           │          │ #0 ödeme X           │
                                   │ #1 ödeme Y           │          │ #1 ödeme Y           │
                                   └──────────────────────┘          └──────────────────────┘
┌──────────────────────┐          ┌──────────────────────┐
│ Partition 2 (follower)│◄────────►│ Partition 2 (leader) │
│ #0 iptal Z           │          │ #0 iptal Z           │
└──────────────────────┘          └──────────────────────┘

Sıra garantisi sadece partition içinde vardır. Farklı partition’lardaki mesajların göreceli sırası garanti değildir. Producer tarafında mesajı bir key ile gönderirsen, aynı key aynı partition’a düşer, sıra korunur.

Redpanda: Kafka’nın C++ Tereyağı

Redpanda, Kafka API uyumlu, ama C++ ile yazılmış, single-binary bir alternatif:

Zookeeper yok: Kafka eski sürümleri ZK’ye bağımlıydı (metadata yönetimi). Redpanda KRaft benzeri iç consensus kullanır.
Düşük latency: tipik p99 latency 5-10 ms (Kafka’ya kıyasla).
Daha az operasyon: tek binary, daha az tuning.
Kafka API uyumlu: aynı client kütüphaneleri, aynı konfigürasyon dosyaları, aynı topic/partition yapısı.

Geliştirme makinende Redpanda’yı 5 dakikada ayağa kaldırırsın:

docker run -d --name redpanda \
  -p 9092:9092 -p 9644:9644 \
  docker.redpanda.com/redpandadata/redpanda:latest \
  redpanda start --overprovisioned --smp 1 --memory 1G \
  --reserve-memory 0M --check=false --node-id 0 \
  --kafka-addr PLAINTEXT://0.0.0.0:9092 \
  --advertise-kafka-addr PLAINTEXT://localhost:9092

Producer: Yapılandırılmış Yazma

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers="localhost:9092",
    key_serializer=lambda k: k.encode("utf-8"),                  # user_id key
    value_serializer=lambda v: json.dumps(v).encode("utf-8"),
    acks="all",                                                  # tüm replica onaylasın
    retries=3,                                                   # ağ hatasında 3 kez dene
    linger_ms=10,                                                # 10ms batch'le
    compression_type="lz4",                                      # mesajları sıkıştır
)

# Aynı müşterinin siparişleri aynı partition'a düşer (sıra korunur)
siparis = {"urun": "limonata", "fiyat": 25, "musteri_id": "u123"}
gelecek = producer.send("siparisler", key="u123", value=siparis)

# Sonucu al (asenkron; meta-data döner)
metadata = gelecek.get(timeout=10)
print(f"partition={metadata.partition} offset={metadata.offset}")

Kritik ayarlar:

acks="all": lider broker mesajı aldığında değil, tüm ISR (in-sync replicas) onayladığında başarılı say. Veri kaybına karşı dayanıklı.
retries=3: geçici ağ hatasında tekrar dene. enable_idempotence=True ile tekrarlanan gönderimde aynı mesajı iki kere yazma garantisi.
linger_ms=10: küçük bir gecikme ile mesajları toplu gönder, throughput artar.
compression_type="lz4": ağ bant genişliğini koru.

Consumer: Güvenli Okuma

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    "siparisler",
    bootstrap_servers="localhost:9092",
    group_id="rapor-servisi",
    key_deserializer=lambda b: b.decode("utf-8"),
    value_deserializer=lambda b: json.loads(b.decode("utf-8")),
    auto_offset_reset="earliest",
    enable_auto_commit=False,                                    # manuel kontrol
    max_poll_records=100,                                        # her poll'da en fazla 100
)

# Manuel offset yönetimi (güvenli)
try:
    for mesaj in consumer:
        try:
            # işle
            process(mesaj.value)
        except Exception as e:
            # hata: log'la ve offset'i commit etme
            print(f"hata: {e}; offset {mesaj.offset} atlanıyor")
            continue
        consumer.commit()                                       # başarılıysa commit
except KeyboardInterrupt:
    pass
finally:
    consumer.close()

enable_auto_commit=False üretim için önerilen: otomatik commit, mesajı işlemeden önce offset’i kaydederse veri kaybı olur. Manuel commit ile “önce işle, sonra commit” garantisi verirsin.

Exactly-Once Semantics (EOS)

Kafka Streams ve Kafka transactional API, uçtan uca exactly-once sağlar:

Producer: enable_idempotence=True + transactional_id ile her mesaj için sequence number.
Consumer: isolation.level="read_committed" ile commit edilmemiş transaction mesajlarını atla.
Sink: Kafka’ya yazıyorsan aynı transaction içinde atomik yazılır.

Klasik üçlü çözüm: idempotent producer + transactional state + read_committed consumer. Flink de Kafka’yı source/sink olarak kullanırken EOS sağlar (two-phase commit ile).

Avro ve Schema Evolution

Avro’nun JSON’dan farkı: şema veriyle birlikte değil, ayrı bir Schema Registry’de tutulur. Mesaj sadece değerleri binary taşır.

Avantajı: şema evrim geçirebilir. Eski producer yeni şemayla yazarsa eski consumer kırılmaz; yeni alanlar default değerle gelir, eski alanlar atlanır.

Schema Registry akışı:

Producer                  Schema Registry                Broker
   │                             │                          │
   │── register schema v1 ──────►│                          │
   │                             │                          │
   │── send message (id=1) ──────┼─────────────────────────►│
   │                             │                          │
Consumer                        │                          │
   │── fetch (id=1) ────────────┼─────────────────────────►│
   │   (lookup schema by id) ──►│                          │
   │◄─ schema v1 ───────────────│                          │

Confluent Schema Registry açık kaynak, Redpanda da uyumlu çalışır. Şema subject (topic adı + key/value son eki) ile versiyonlanır:

# Yeni şema kaydet
from schema_registry.client import SchemaRegistryClient
from schema_registry.schema import Schema

sr = SchemaRegistryClient(url="http://localhost:8081")

sema = Schema(
    schema_str=open("siparis.v1.avsc").read(),
    schema_type="AVRO",
)

schema_id = sr.register("siparisler-value", sema)
print(f"kayıt no: {schema_id}")

Uyumluluk Modları

Backward: yeni şema eski consumer’ı kırmaz (eski alanlar yeni şemada da var).
Forward: yeni şema eski producer’ı kırmaz.
Full: hem backward hem forward.
Backward en yaygın — yeni şema eklenince eski okuyucular hâlâ çalışır.

KSQL / Kafka Streams: SQL ve Java

Kafka üstünde akan SQL için KSQL (veya ksqlDB) açık kaynak:

-- Akan sorgu: her müşterinin son 5 dakikadaki sipariş sayısı
CREATE TABLE musteri_siparis_5dk AS
  SELECT musteri_id,
         COUNT(*) AS siparis_sayisi,
         LATEST_BY_OFFSET(urun) AS son_urun
  FROM siparisler
  WINDOW TUMBLING (SIZE 5 MINUTES)
  GROUP BY musteri_id
  EMIT CHANGES;

Java ekosisteminde ise Kafka Streams DSL’i ile stateful dönüşümler:

KStream<String, Siparis> stream = builder.stream("siparisler");
KTable<String, Long> sayilar = stream
    .groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
    .count();
sayilar.toStream().to("musteri-bazli-siparis-sayisi");

Bu ders KSQL / Kafka Streams kurulumu yapmıyor; bil ki var.

Özet

Bugün öğrendik:

Kafka = dağıtılmış, bölümlenmiş, replikalı commit log. Sıra sadece partition içinde.
Redpanda Kafka API uyumlu, tek binary, düşük latency C++ alternatif.
Producer ayarları: acks=all, enable_idempotence, linger_ms, compression.
Consumer ayarları: group_id, enable_auto_commit=False, manuel commit.
Exactly-once = idempotent producer + transactional sink + read_committed consumer.
Avro + Schema Registry: şema evrimi ile geriye/ileriye uyumlu mesaj formatı.
KSQL / Kafka Streams: akan SQL veya stateful stream API (opsiyonel bonus).

Alıştırma

Redpanda’yı Docker’da ayağa kaldır, basit bir producer/consumer çalıştır.
JSON yerine Avro ile mesaj yaz; mesaj boyutunu karşılaştır.
acks="all" yerine acks=0 dene. Hangi risk ortaya çıkıyor?
Bir cümle yaz: “Neden auto_offset_reset='latest' dikkatli kullanılmalı?”

Sıradaki

Günlüğümüz hazır, akan veriyi işleyip bir yere indirelim: PyFlink + Postgres sink ile kelime sayacı →