Bagian 1: Memahami Event Stream

Pipeline Airflow pada proyek sebelumnya memperlakukan data sebagai kumpulan baris yang diolah sekali jalan. Streaming membalik asumsi tersebut: data datang terus-menerus sebagai deretan event — fakta yang sudah terjadi, bertimestamp, dan tidak pernah diubah setelah diterbitkan. Sebelum menulis producer maupun consumer, kita harus memutuskan dengan tepat apa isi satu event pesanan TokoKita, karena keputusan ini menentukan desain topic pada Bagian 5, aturan validasi pada Bagian 8, hingga cara memantau stream pada Bagian 9. Bagian ini merancang bentuk event dan membangkitkan dataset deterministik yang dipakai seluruh bagian berikutnya.

Dari Batch Terjadwal ke Event Stream #

Pada batch pipeline, unit kerja adalah tabel berukuran tetap; pada streaming, unit kerjanya adalah satu event yang diterbitkan begitu peristiwa terjadi. Chapter Konsep Stream Processing dan Batch vs Stream Processing membedah perbedaan keduanya; tabel berikut merangkum konsekuensi praktisnya bagi proyek ini:

Baris “kegagalan di tengah jalan” adalah inti proyek ini: ketika consumer berhenti setengah jalan, Kafka mengirim ulang pesan yang belum di-commit, sehingga hilir harus siap menerima duplikat — masalah yang kita selesaikan pada Bagian 6 dan 7.

Anatomi Event Pesanan #

Satu event mewakili satu peristiwa checkout: pesanan dibuat, dibayar, dikirim, selesai, atau dibatalkan. Setiap event membawa identitas, versi schema, waktu kejadian, dan muatan pesanan:

Ketiga event pertama dari simulator menunjukkan bentuk final yang dikirim sebagai nilai pesan Kafka (JSON UTF-8):

from src.producer import buat_event_pesanan, serialisasi_event

for nomor in (1, 2, 6):
    event = buat_event_pesanan(nomor, seed=42)
    print(f"--- event #{nomor} ---")
    print(serialisasi_event(event).decode("utf-8"))

Output yang Dihasilkan:

--- event #1 ---
{"event_id":"evt-000001","schema_version":1,"event_time":"2024-03-01T09:00:09+00:00","pesanan_id":"ORD-1001","pelanggan_id":"PLG-115","wilayah":"makassar","total":2362500,"status":"dibatalkan"}
--- event #2 ---
{"event_id":"evt-000002","schema_version":1,"event_time":"2024-03-01T09:00:12+00:00","pesanan_id":"ORD-1002","pelanggan_id":"PLG-191","wilayah":"jakarta","total":1043000,"status":"selesai"}
--- event #6 ---
{"event_id":"evt-000006","schema_version":1,"event_time":"2024-03-01T09:01:04+00:00","pesanan_id":"ORD-1006","pelanggan_id":"PLG-130","wilayah":"jakarta","total":1462000,"status":"dibayar"}

Pembangkitan Event Deterministik #

Seluruh demo proyek harus menghasilkan output yang sama setiap kali dijalankan, jadi pembangkit event tidak boleh bergantung pada jam mesin. Fungsi buat_event_pesanan di src/producer.py menurunkan pembangkit acak dari pasangan (seed, nomor):

def buat_event_pesanan(
    nomor: int,
    seed: int = 42,
    mulai_waktu: datetime = MULAI_EVENT,
    bobot_wilayah: list[int] | None = None,
) -> dict:
    """Bangun satu event pesanan deterministik per pasangan ``(seed, nomor)``.

    Pembangkit acak dibuat ulang dari ``seed`` dan ``nomor`` sehingga event
    nomor tertentu selalu identik, apa pun batas batch pemanggilnya —
    properti penting agar hasil produksi bisa direproduksi persis.
    """
    rng = random.Random(f"{seed}:{nomor}")
    wilayah = rng.choices(WILAYAH, weights=bobot_wilayah or BOBOT_WILAYAH)[0]
    event_time = mulai_waktu + timedelta(seconds=(nomor - 1) * 12 + rng.randint(0, 9))
    return {
        "event_id": f"evt-{nomor:06d}",
        "schema_version": 1,
        "event_time": event_time.isoformat(),
        "pesanan_id": f"ORD-{1000 + nomor}",
        "pelanggan_id": f"PLG-{rng.randrange(1, 200):03d}",
        "wilayah": wilayah,
        "total": rng.randrange(25_000, 2_500_000, 500),
        "status": rng.choice(STATUS_PESANAN),
    }

Mari kita telaah baris-baris kode di atas (nomor mengikuti posisi di src/producer.py):

1. Baris 36 (rng = random.Random(...)): pembangkit acak dibuat dari string "{seed}:{nomor}" — bukan dari satu seed untuk seluruh batch — sehingga event nomor 7 identik baik dibangkitkan sendirian maupun sebagai bagian batch 100 event. Modul random stdlib menjamin hasil yang sama untuk seed yang sama.
2. Baris 37 (wilayah = ...): wilayah dipilih dengan bobot BOBOT_WILAYAH (jakarta 70, bandung 10, surabaya 8, medan 7, makassar 5) — distribusi timpang yang sengaja dijaga untuk demo hot partition pada Bagian 5.
3. Baris 38 (event_time = ...): event time bergerak ±12 detik per nomor event plus jitter 0-9 detik, meniru checkout yang datang berdekatan; lima event rata-rata mengisi satu jendela menit.
4. Baris 40-42 (event_id, schema_version, event_time): identitas event, versi kontrak, dan stempel waktu event — tiga field infrastruktur yang dipisahkan dari muatan bisnis.
5. Baris 43-47 (pesanan_id s.d. status): muatan bisnis pesanan; total dijangkarkan kelipatan 500 antara 25 ribu hingga 2,5 juta rupiah, dan status diambil dari lima nilai tetap yang juga menjadi enum pada schema Bagian 8.

Satu Pesanan, Banyak Event #

Karakteristik immutable sering disalahpahami sebagai “data pesanan tidak boleh berubah”. Yang tidak berubah adalah event-nya, bukan pesanannya. Satu pesanan ORD-1001 bergerak melalui beberapa peristiwa, dan setiap peristiwa menjadi event tersendiri yang mengikat pesanan_id sama. Sebagai ilustrasi konseptual (bukan keluaran simulator, yang menghasilkan event acak per pesanan), perjalanan satu pesanan dapat tercatat demikian:

Urutan empat event inilah yang harus terjaga — dan karena semuanya berkunci pelanggan_id yang sama, keempatnya jatuh pada satu partition sehingga urutannya dijamin. Hilir yang ingin tahu “status terkini pesanan” cukup membaca event terakhir per pesanan_id; hilir yang ingin tahu “perjalanan pesanan” membaca semuanya. Kedua kebutuhan terlayani oleh satu stream yang sama — inilah keunggulan event log dibanding tabel status yang ditimpa.

Karakteristik yang Kita Pegang #

Tiga karakteristik event stream dari chapter Konsep Stream Processing diterjemahkan menjadi keputusan konkret proyek ini. Pertama, event bersifat immutable: sekali diterbitkan, isi evt-000001 tidak pernah ditimpa; perubahan status pesanan menjadi event baru dengan pesanan_id sama. Kedua, stream bersifat unbounded: consumer tidak tahu kapan stream berakhir, sehingga pola baca selalu poll berulang — bukan satu query. Ketiga, event dapat diputar ulang: selama retensi topic masih menampung pesan, consumer group baru bisa membaca histori dari offset 0, kemampuan yang kita manfaatkan untuk audit pada Bagian 6.

Dengan bentuk event yang tetap dan dataset yang dapat direproduksi, kita siap menyiapkan media penyimpanannya: topic Kafka pada cluster KRaft, yang kita konfigurasikan pada bagian berikutnya.

Referensi resmi: dokumentasi Apache Kafka .