Stream vs Table #

Dalam dunia pemrosesan data real-time, tantangan terbesar kita sering kali adalah menyelaraskan dua jenis informasi yang berbeda secara struktural: sejarah transaksi yang mengalir terus-menerus dan status terkini dari entitas bisnis kita. Sebagai contoh, di sistem perbankan, kita memiliki aliran riwayat transaksi transfer uang (yang terjadi setiap detik) dan kita juga memiliki tabel saldo rekening pengguna saat ini. Apache Kafka memecahkan tantangan ini dengan memperkenalkan konsep Stream-Table Duality (Dualitas Aliran dan Tabel) yang diwujudkan melalui abstraksi pemrograman KStream, KTable, dan GlobalKTable. Melalui artikel ini, kita akan membedah secara mendalam perbedaan fundamental antara aliran data dan tabel, cara kerja masing-masing abstraksi di Kafka Streams, use case spesifik untuk ketiganya, serta bagaimana melakukan transformasi data di antara tipe-tipe tersebut secara aman tanpa memicu kebocoran memori atau ketidaksesuaian partisi.

Konsep Fundamental Stream-Table Duality #

Sebelum kita masuk ke ranah kode dan arsitektur internal, kita harus memahami dasar teoretis yang sangat elegan ini: sebuah aliran data (stream) dapat diubah menjadi tabel (table), dan sebuah tabel dapat diubah kembali menjadi aliran data.

1. Hubungan Dua Arah yang Erat #

Dualitas ini bekerja dalam dua arah yang saling melengkapi dan membentuk dasar komputasi stateful:

• Stream ke Table (Agregasi): Aliran data merepresentasikan sejarah masa lalu hingga saat ini. Setiap record baru adalah kejadian independen. Jika kita memutar ulang (replay) seluruh event di dalam sebuah aliran data dari awal hingga akhir dan mengagregasikannya berdasarkan kunci tertentu, kita akan menghasilkan sebuah tabel keadaan (state table). Ini mirip dengan bagaimana buku kas akuntansi (aliran transaksi debit/kredit) menghasilkan saldo akhir buku besar (tabel). Tanpa adanya aliran data asli, tabel keadaan tersebut tidak bisa dibentuk dari nol.
• Table ke Stream (Changelog): Sebuah tabel merepresentasikan keadaan saat ini (latest state). Jika kita memantau setiap modifikasi (penambahan, perubahan, penghapusan) pada sebuah tabel dan memancarkan perubahan tersebut sebagai event baru, kita akan menghasilkan aliran data log perubahan (changelog stream). Di database tradisional, ini setara dengan Transaction Log (seperti WAL di PostgreSQL atau Binlog di MySQL) yang memancarkan perubahan baris ke sistem replikasi atau CDC (Change Data Capture).

2. Implemetasi Logis dalam Kafka #

Di dalam Kafka, dualitas ini sangat nyata. Sebuah topik Kafka secara fisik adalah log kejadian (stream). Namun, dengan menggunakan konfigurasi Log Compaction (cleanup.policy=compact), Kafka mempertahankan hanya record terbaru untuk setiap kunci. Ketika sebuah aplikasi Kafka Streams membaca topik terkompresi tersebut sebagai KTable, ia sedang merekonstruksi state table tersebut ke dalam database RocksDB lokal dari log stream asli yang disimpan di broker.

flowchart LR
    subgraph STREAMS ["KStream (Append-Only Events)"]
        direction LR
        E1["UserA: +$100"] --> E2["UserA: -$30"]
        E2 --> E3["UserB: +$50"]
        E3 --> E4["UserA: +$10"]
    end

    subgraph DUALITY ["Dualitas / Transformasi"]
        direction TB
        Agg["Agregasi (Replay & GroupBy Key)"]
        Change["Changelog (Emit State Changes)"]
    end

    subgraph TABLES ["KTable (Upsert State Table)"]
        direction TB
        State1["UserA: Saldo $80"]
        State2["UserB: Saldo $50"]
    end

    STREAMS -->|"GroupBy & Aggregate"| Agg
    Agg --> TABLES
    TABLES -->|"Emit Changes"| Change
    Change --> STREAMS

    style E1 stroke:#0288d1,stroke-width:2px
    style E2 stroke:#0288d1,stroke-width:2px
    style E3 stroke:#0288d1,stroke-width:2px
    style E4 stroke:#0288d1,stroke-width:2px
    style State1 stroke:#388e3c,stroke-width:2px
    style State2 stroke:#388e3c,stroke-width:2px

KStream: Aliran Peristiwa Append-Only #

KStream adalah abstraksi dari aliran data peristiwa biasa. Setiap record data yang masuk ke dalam KStream diperlakukan sebagai entitas mandiri yang bersifat append-only (hanya ditambahkan di akhir log).

Karakteristik Utama KStream #

• Semantik Fakta: Data di dalam KStream bersifat imut (immutable). Sekali sebuah peristiwa terjadi dan dicatat ke broker Kafka, peristiwa tersebut adalah fakta sejarah. Kita tidak bisa menghapus atau memodifikasi record tersebut. Yang bisa kita lakukan hanyalah mengirimkan event baru untuk membatalkan atau merevisi dampak event lama.
• Penanganan Kunci Duplikat: Jika ada dua record data dengan kunci (key) yang sama masuk ke KStream, keduanya dianggap sebagai dua peristiwa yang berbeda dan independen. Keduanya akan diproses satu per satu secara berurutan. Misalnya, jika pengguna menekan tombol “Beli” sebanyak dua kali, KStream akan mencatat dua transaksi terpisah, bukan memperbarui transaksi pertama.
• Contoh Kasus Penggunaan:
  • Peta koordinat GPS dari armada pengiriman yang dikirimkan setiap 5 detik.
  • Clickstream pengguna di situs web e-commerce untuk analisis corong konversi.
  • Log akses server web untuk pemantauan keamanan real-time.
  • Sinyal sensor suhu mesin pabrik untuk pemeliharaan prediktif.

KTable: Tabel Changelog Stateful #

KTable adalah abstraksi dari tabel keadaan lokal. Berbeda dengan KStream, setiap record data yang masuk ke dalam KTable diperlakukan sebagai sebuah pembaruan (upsert - update jika ada, insert jika belum ada) berdasarkan kuncinya.

Karakteristik Utama KTable #

• Semantik Keadaan (State): Data di dalam KTable merepresentasikan status terakhir dari suatu entitas pada titik waktu tertentu. Kita menggunakan KTable ketika kita hanya peduli pada nilai terakhir dari suatu kunci.
• Penanganan Kunci Duplikat: Jika ada record baru dengan kunci yang sudah ada di KTable, record lama akan ditimpa (overwrite) dengan nilai yang baru di memori lokal. RocksDB yang menjadi mesin penyimpanan lokal KTable akan melakukan pembaruan entri key-value tersebut secara instan.
• Tombstone (Penghapusan): KTable mendukung penghapusan data secara eksplisit. Jika kita mengirimkan record dengan kunci tertentu namun dengan nilai (value) bernilai null (disebut sebagai tombstone marker), KTable akan menghapus kunci tersebut dari tabel lokalnya dan memancarkan event penghapusan tersebut ke sistem hilir.
• Contoh Kasus Penggunaan:
  • Saldo rekening berjalan dari nasabah bank.
  • Status keaktifan kurir (apakah sedang online, sibuk, atau offline).
  • Profil detail pelanggan e-commerce (seperti alamat pengiriman saat ini atau jenis membership).
  • Harga saham terkini di bursa efek.

GlobalKTable: Tabel Replikasi Global #

GlobalKTable mirip dengan KTable biasa karena keduanya menggunakan semantik upsert untuk menyimpan keadaan terbaru. Namun, perbedaan mendasar mereka terletak pada bagaimana data didistribusikan ke seluruh instansi aplikasi kita dan bagaimana konsumsi memori dikelola.

Perbedaan Arsitektur KTable vs GlobalKTable #

• KTable (Partisi Terdistribusi): KTable dipartisi secara lokal sesuai dengan pembagian partisi topik Kafka. Jika topik input memiliki 4 partisi dan kita memiliki 4 StreamTask yang berjalan di 2 instansi aplikasi berbeda, masing-masing instansi hanya akan memuat data dari 2 partisi (50% dari total data).
  • Kelebihan: Skalabilitas horizontal yang sangat baik. Jika volume data membengkak, kita cukup menambah instansi aplikasi baru untuk membagi beban memori RocksDB.
  • Kekurangan: Membutuhkan aturan co-partitioning yang ketat jika kita ingin men-join-kan KStream dengan KTable. Kunci transaksi pada KStream dan kunci profil pada KTable harus dipartisi dengan cara yang sama di broker.
• GlobalKTable (Replikasi Penuh): GlobalKTable memuat 100% data dari topik masukan ke dalam setiap instansi aplikasi yang berjalan, tidak peduli berapa partisi yang dimiliki topik input dan berapa jumlah task yang ditugaskan ke instansi tersebut.
  • Kelebihan: Sangat memudahkan proses join. Kita dapat men-join-kan KStream apa saja dengan GlobalKTable tanpa perlu melakukan partisi ulang (re-partitioning), karena data penyeimbang pasti tersedia di memori lokal instansi saat itu juga.
  • Kekurangan: Mengkonsumsi resource memori RAM (JVM heap) dan ruang disk (RocksDB) jauh lebih besar karena duplikasi penuh. Jika data dalam topik GlobalKTable melebihi kapasitas memori satu VM server, aplikasi akan mengalami OutOfMemoryError. Oleh karena itu, GlobalKTable hanya cocok untuk tabel referensi statis berukuran kecil (seperti tabel kode pos, daftar kategori produk, atau tabel mata uang).

Kode Penerapan Java: KStream, KTable, dan GlobalKTable #

Mari kita lihat contoh kode Java untuk menginisialisasi ketiga jenis abstraksi ini menggunakan Kafka Streams DSL API.

Skenario Implementasi #

Kita ingin membangun sistem pemrosesan transaksi pembayaran e-commerce. Kita memiliki:

1. Topik payment-transactions berisi aliran transaksi (KStream).
2. Topik user-profiles berisi profil dan tingkat keanggotaan pengguna yang bisa berubah sewaktu-waktu (KTable).
3. Topik currency-rates berisi nilai tukar mata uang asing statis (GlobalKTable).

// BENAR: Menginisialisasi dan menghubungkan KStream, KTable, dan GlobalKTable di Java SDK.
// ✓ Mematuhi aturan co-partitioning untuk KTable lokal dan replikasi penuh untuk GlobalKTable.
public class StreamTableDualityExample {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();

        // 1. Inisialisasi KStream (Append-Only Log)
        // ✓ Membaca setiap transaksi sebagai event mandiri
        KStream<String, String> transactionStream = builder.stream(
            "payment-transactions",
            Consumed.with(Serdes.String(), Serdes.String())
        );

        // 2. Inisialisasi KTable (State Table - Lokal Partisi)
        // ✓ Secara otomatis menyimpan keadaan terbaru profil pengguna per partisi lokal
        // RocksDB lokal akan menyimpan data ini dan memutakhirkan key-value jika ada profil baru
        KTable<String, String> userProfileTable = builder.table(
            "user-profiles",
            Consumed.with(Serdes.String(), Serdes.String())
        );

        // 3. Inisialisasi GlobalKTable (Global Replicated Table)
        // ✓ Seluruh data kurs mata uang direplikasi ke semua instansi tanpa peduli partisi
        GlobalKTable<String, String> currencyRateTable = builder.globalTable(
            "currency-rates",
            Consumed.with(Serdes.String(), Serdes.String())
        );

        // Contoh Logika Bisnis: Filtering Transaksi Sederhana
        transactionStream
            .filter((key, value) -> value != null && value.contains("SUCCESS"))
            .to("successful-payments");

        Topology topology = builder.build();
    }
}

Transformasi: Mengubah KStream Menjadi KTable (dan Sebaliknya) #

Salah satu kekuatan utama Kafka Streams adalah fleksibilitasnya untuk bertransisi di antara kedua model ini secara dinamis.

Anti-Pattern: Menggunakan Cache Manual atau Thread Eksternal #

Mencoba melakukan agregasi dari aliran KStream menjadi tabel status menggunakan variabel global Java (HashMap statis) atau threads scheduler manual adalah kesalahan fatal yang sering dilakukan developer pemula.

// ANTI-PATTERN: Melakukan kalkulasi stateful KStream ke KTable secara manual dengan HashMap lokal.
// ✗ State akan hilang saat instansi aplikasi crash, memicu inkonsistensi memori, dan melanggar paralelisme thread.
public class ManualStateAggregation {
    // ✗ BAHAYA: HashMap statis tidak thread-safe dan tidak di-backup ke broker (disaster recovery gagal!)
    private static final Map<String, Double> userBalances = new ConcurrentHashMap<>();

    public static void processStreamManual(KStream<String, String> stream) {
        stream.foreach((userId, transactionJson) -> {
            double amount = parseAmount(transactionJson);
            
            // ✗ Modifikasi manual memori internal tanpa koordinasi state store transaksional RocksDB
            userBalances.merge(userId, amount, Double::sum);
            
            log.info("User {} balance updated to {}", userId, userBalances.get(userId));
        });
    }

    private static double parseAmount(String json) { return 100.0; }
}

Mengapa Pola Manual di Atas Sangat Berbahaya? #

1. Kehilangan Data Saat Crash: HashMap statis hanya hidup di memori RAM instance JVM. Jika Kubernetes me-restart container aplikasi kita karena kegagalan liveness probe, seluruh data saldo berjalan tersebut akan hilang selamanya.
2. Race Conditions: Ketika kita menaikkan jumlah thread melalui konfigurasi num.stream.threads, beberapa thread pemroses data akan mengakses HashMap yang sama secara paralel, memicu persaingan memori.
3. Ketidakmampuan Scaling: Jika kita mendeploy instance kedua dari microservice ini, instance baru tersebut tidak akan memiliki akses ke HashMap di instance pertama. Data saldo akan terpecah tanpa sinkronisasi.

Solusi Praktis: Menggunakan Agregasi Stateful API #

Kafka Streams menyediakan operator DSL terkelola untuk mereduksi atau mengagregasikan KStream menjadi KTable secara aman. State store akan didukung oleh RocksDB lokal dan topik Changelog secara otomatis di bawah kap mesin.

// BENAR: Mengubah KStream menjadi KTable melalui operasi GROUP BY dan AGGREGATE.
// ✓ Aman dari crash, terintegrasi dengan RocksDB, dan ter-backup penuh ke topik changelog internal.
public class StreamToTableDuality {
    public static KTable<String, Double> aggregateStream(KStream<String, String> transactionStream) {
        
        return transactionStream
            // Langkah A: Kelompokkan aliran data berdasarkan User ID (Key)
            // ✓ Operasi ini memicu internal repartition jika key dirubah, memastikan co-partitioning.
            .groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
            
            // Langkah B: Agregasikan nilai transaksi ke dalam State Store lokal
            .aggregate(
                // 1. Initializer: Setel saldo awal menjadi 0.0
                () -> 0.0,
                
                // 2. Aggregator: Tambahkan jumlah transaksi baru ke saldo berjalan
                (userId, transactionJson, currentBalance) -> {
                    double amount = parseAmount(transactionJson);
                    return currentBalance + amount;
                },
                
                // 3. Materialized: Konfigurasikan RocksDB state store bernama "user-balance-store"
                // RocksDB lokal akan menyimpan state, dan Kafka Streams secara asinkron membackup
                // setiap perubahan ke topik changelog internal bernama: application-id-user-balance-store-changelog
                Materialized.<String, Double, KeyValueStore<Bytes, byte[]>>as("user-balance-store")
                    .withKeySerde(Serdes.String())
                    .withValueSerde(Serdes.Double())
            ); // ✓ Output dari aggregate() adalah KTable<String, Double>
    }

    private static double parseAmount(String json) {
        // Parsing logika
        return 10.0;
    }
}

Jika kita ingin melakukan sebaliknya (mengubah KTable kembali menjadi KStream), kita cukup memanggil fungsi toStream() pada objek KTable kita:

KTable<String, Double> userBalances = aggregateStream(transactionStream);

// Mengubah KTable saldo berjalan menjadi KStream (Changelog stream) untuk dikirim ke topik eksternal
KStream<String, Double> balanceChangelogStream = userBalances.toStream();
balanceChangelogStream.to("user-balance-updates", Produced.with(Serdes.String(), Serdes.Double()));

Dengan memanggil toStream(), setiap kali saldo seorang pengguna diperbarui di RocksDB, KTable akan memancarkan satu record perubahan yang memuat User ID dan Saldo Baru ke topik tujuan user-balance-updates.

Masalah Umum dan Cara Mengatasinya (Troubleshooting) #

Ketika kita menggunakan KTable dan GlobalKTable di lingkungan produksi, ada beberapa kendala umum yang harus kita antisipasi secara preventif:

1. Kebocoran Memori (Out of Memory) pada GlobalKTable #

• Masalah: Topik yang dibaca sebagai GlobalKTable terus bertambah datanya seiring waktu tanpa batas, menyebabkan memori heap JVM di setiap instance habis.
• Solusi: Pastikan topik Kafka yang mendasari GlobalKTable dikonfigurasi dengan kebijakan pembersihan terkompresi (cleanup.policy=compact) dan memiliki batasan retensi ukuran segmen yang ketat. Selalu monitor metrik JVM memory usage di JVM dashboard kita.

2. Kesenjangan Sinkronisasi (Sync Gap) RocksDB #

• Masalah: Saat aplikasi di-restart di node Kubernetes baru, RocksDB membutuhkan waktu lama untuk melakukan replay data dari topik changelog Kafka sebelum siap melayani transaksi (cold start latency).
• Solusi: Aktifkan fitur Standby Replicas dengan menyetel konfigurasi num.standby.replicas=1. Ini akan memerintahkan instance lain untuk menduplikasi changelog secara pasif sehingga siap mengambil alih kepemilikan task secara instan tanpa proses restore yang lama.

Matriks Perbandingan: KStream vs KTable vs GlobalKTable #

Untuk mempermudah pemilihan tipe data saat merancang arsitektur, berikut adalah rangkuman perbandingan ketiganya:

Ringkasan #

• Stream-Table Duality — Hubungan dua arah di mana aliran peristiwa (KStream) dapat diakumulasikan menjadi tabel status (KTable), dan tabel status dapat dipantau perubahannya menjadi aliran peristiwa kembali.
• KStream — Abstraksi data append-only di mana setiap record baru dianggap sebagai fakta sejarah yang tidak mengubah record sebelumnya.
• KTable — Abstraksi data stateful di mana record baru dengan key yang sama akan menimpa record lama (upsert), serta mendukung penghapusan via record null (tombstone).
• GlobalKTable — Abstraksi tabel data di mana 100% data direplikasikan ke seluruh instansi aplikasi, membebaskan kebutuhan join dari syarat co-partitioning dengan konsekuensi konsumsi memori tinggi.
• Tombstone Marker — Event dengan value null yang digunakan di KTable/GlobalKTable untuk menghapus key terkait dari RocksDB local state store secara permanen.
• toStream() — Metode instan untuk mengekspos transisi state dari KTable kembali menjadi aliran changelog event (KStream) agar bisa dikirimkan ke topik Kafka lain.

← Sebelumnya: Apa itu Kafka Streams?   Berikutnya: Topology vs Processor →