KAL
Menu

Primitif stream

Operasi stateless pada KStream: map, filter, flatMap, join, dan groupBy beserta semantiknya.

Daftar isi

Operasi stateless mengolah setiap record berdasarkan isi record itu sendiri, tanpa menyimpan keadaan apa pun antar record. Topik ini membahas lima primitif yang paling sering digunakan pada KStream, yaitu filter, map, flatMapValues, join berwindow, dan groupBy beserta semantiknya, lengkap dengan potongan Java untuk tiap operasi. Satu simulasi Python memutar pipeline flatMap, filter, map, dan groupBy pada daftar kejadian tetap, sehingga keluaran tiap tahap dapat diperiksa. Setelah membacanya, batas antara transformasi tanpa state dan operasi yang mulai menyimpan state dapat ditetapkan dengan sadar.

Record pada KStream dan sifat operasi stateless#

Setiap record pada KStream membawa tiga bagian, yaitu key, value, dan timestamp. Operator stateless menentukan keluarannya hanya dari record yang sedang diproses, sehingga dua task yang memproses record yang sama menghasilkan keluaran yang sama tanpa saling berkomunikasi. Karena tidak ada keadaan yang dipegang, memulai ulang task cukup dengan membaca topic kembali dari offset tertentu, dan pemulihannya tidak menyentuh state store sama sekali.

Operator stateless dirangkai menjadi topologi berantai. Setiap pemanggilan mengembalikan KStream baru yang mewakili node berikutnya pada graf, sehingga topologi sumber-filter-sink tersusun dari rantai pemanggilan yang dibaca seperti kalimat. Urutan perangkaiannya bebas, tetapi urutan semantiknya menentukan hasil, misalnya menyaring sebelum mengubah nilai membuat record yang akhirnya dibuang tidak pernah diformat.

Batas antara stateless dan stateful jelas pada DSL Kafka Streams. Operator filter, map, mapValues, dan flatMapValues tidak menyimpan apa pun, sedangkan join, windowing, dan agregasi seperti count menambahkan state store beserta changelog topic-nya. Bagian akhir topik ini menyentuh groupBy persis pada batas tersebut karena menjadi gerbang menuju agregasi.

Operasi filter: menyaring record#

Semantik filter adalah yang paling sederhana di antara primitif lainnya. Setiap record diuji dengan satu predikat, yaitu fungsi yang mengembalikan benar atau salah, dan hanya record dengan predikat bernilai benar yang lolos. Record yang lolos diteruskan tanpa perubahan apa pun pada key maupun value-nya, sedangkan record yang dibuang berhenti di node itu selamanya. Jumlah record keluaran ditentukan sepenuhnya oleh isi record masukan karena tidak ada keadaan yang dipegang.

Potongan Java berikut melanjutkan topologi suhu pada topik Kafka Streams, yaitu menahan bacaan di bawah 30 derajat sebelum diteruskan ke node berikutnya. Lambda-nya menerima key dan value dari tiap record, lalu mengembalikan nilai predikatnya.

java Contoh: filter menahan bacaan di bawah 30 derajat 
// tidak-dijalankan
KStream<String, Double> panas = suhu.filter(
    (sensor, nilai) -> nilai >= 30.0);

Penyaringan beruntun setara dengan satu predikat gabungan, sehingga dua pemanggilan filter berurutan menghasilkan keluaran yang sama dengan satu filter yang predikatnya digabung. Pemisahan tetap berguna karena predikat kecil yang bernama jelas lebih mudah diuji daripada predikat besar yang menyatukan semua syarat sekaligus. Biaya tambahannya hanyalah satu pemeriksaan predikat per record, yang jarang menjadi hambatan sebelum throughput menuntut pengukuran.

Operasi map dan mapValues: mengubah key dan nilai#

mapValues mengubah value tiap record dan membiarkan key serta timestamp-nya utuh. map mencakup lebih banyak karena fungsi pemetakannya mengembalikan pasangan key dan value yang baru, sehingga keduanya boleh diganti. Perbedaannya bukan sekadar kelengkapan API, melainkan menyangkut letak partisi, karena key menentukan partisi tempat record berada.

Potongan berikut memformat nilai suhu menjadi teks dengan mapValues, lalu membalik pasangan key dan value dengan map untuk kebutuhan pemrosesan lain.

java Contoh: mapValues mengubah nilai, map mengubah key dan nilai 
// tidak-dijalankan
KStream<String, String> keterangan = suhu.mapValues(
    nilai -> String.format("%.1f C", nilai));        // key tetap

KStream<Double, String> dibalik = suhu.map(
    (sensor, nilai) -> KeyValue.pair(nilai, sensor)); // key dan nilai baru

Keluaran mapValues tetap berada pada partisi yang sama karena key-nya tidak berubah, sedangkan keluaran map berpotensi menyebar ke partisi mana pun sesuai key barunya. Karena itu operasi berbasis key pada tahap berikutnya, misalnya join atau agregasi, perlu menyatukan record ber-key sama ke satu partisi terlebih dahulu. Mekanisme penyatuan itu dibahas pada peringatan di bawah dan pada bagian groupBy.

Operasi flatMapValues: satu record menjadi banyak#

flatMapValues menjalankan fungsi yang mengembalikan banyak nilai, lalu setiap nilai menjadi record baru dengan key dan timestamp yang mengikuti record asalnya. Satu record masukan bisa menghasilkan nol, satu, atau beberapa record keluaran, dan operasi ini paling sering digunakan untuk memecah teks menjadi kata. Fungsi pemecahnya murni menafsirkan value yang diterima, sehingga operasi ini tetap masuk kategori stateless. Varian flatMap yang lebih longgar dapat mengubah key sekaligus, dengan konsekuensi repartisi yang sama seperti map.

Potongan berikut memecah setiap baris catatan menjadi kata, dengan key baris yang tetap terbawa pada setiap kata keluarannya.

java Contoh: flatMapValues memecah baris menjadi kata 
// tidak-dijalankan
KStream<String, String> kata = catatan.flatMapValues(
    baris -> Arrays.asList(baris.toLowerCase().split(" ")));

flatMapValues perlu dibedakan dari mapValues, yang selalu menghasilkan tepat satu record per record masukan. Apabila fungsi pemecah mengembalikan daftar kosong, record asal hilang dari aliran tanpa pesan apa pun. Perilaku ini diinginkan saat penyaringan kata kosong, tetapi menjadi sumber kebingungan bila daftar kosong itu muncul karena bug. Timestamp setiap record turunan juga mengikuti record asalnya, sehingga event-time pada pemrosesan berikutnya tidak bergeser akibat pemecahan. Jumlah record keluaran flatMapValues sebaiknya diperiksa pada data awal sebelum topologi digunakan luas.

Join dua KStream dengan window#

join menyatukan record dari dua KStream yang berbagi key, tetapi kedatangan record kiri dan record kanan tidak dapat dijamin bersamaan, sehingga join stream-stream selalu berpasangan dengan window. Window membatasi rentang waktu maksimum antara kedua record yang masih boleh dipasangkan, misalnya bacaan suhu dan indikator kelembapan dari topic berbeda yang berjarak paling banyak 30 detik. Tanpa window, setiap record kiri harus disimpan selamanya menunggu pasangannya, dan itulah sebabnya join menambahkan state berwindow ke topologi.

Potongan berikut memasangkan bacaan dengan indikator yang tiba paling lambat 30 detik kemudian, dan masa tenggang lima detik menunda penutupan window untuk record yang tiba terlambat.

java Contoh: join berwindow antara dua KStream 
// tidak-dijalankan
KStream<String, String> gabungan = bacaan.join(
    indikator,
    (nilaiBacaan, nilaiIndikator) -> nilaiBacaan + " (" + nilaiIndikator + ")",
    JoinWindows.ofTimeDifferenceAndGrace(
        Duration.ofSeconds(30), Duration.ofSeconds(5)),
    StreamJoined.with(Serdes.String(), Serdes.String(), Serdes.String()));

Fungsi penggabungnya menerima value kiri dan value kanan, lalu menghasilkan value pasangannya, sedangkan key pasangan diambil dari key yang sama pada kedua stream. Record yang tidak menemukan pasangan dalam window tidak menghasilkan apa pun, dan state window-nya dibuang setelah masa tenggang berlalu. Jumlah kombinasi yang dipasangkan tumbuh seiring luas window, sehingga window yang lebih besar menuntut state yang lebih besar pula, seperti dibahas pada topik Windowing.

Operasi groupBy: gerbang menuju agregasi#

groupBy menetapkan key baru untuk setiap record melalui fungsi pemetaan, lalu hasilnya berupa KGroupedStream yang siap diagregasi dengan count, reduce, atau aggregate. Key baru menentukan partisi tujuan, sehingga groupBy mengaktifkan mekanisme repartisi yang sama seperti map. Setelah record ber-key sama terkumpul pada satu partisi, agregasi dapat menggabungkannya menjadi satu nilai per key.

Potongan berikut menjadikan kata sebagai key, lalu menghitungnya dengan count. Hasilnya berupa KTable, yaitu keadaan terkini per kata, bukan KStream.

java Contoh: groupBy lalu count menghasilkan KTable 
// tidak-dijalankan
KTable<String, Long> jumlahKata = kata.groupBy(
        (kunci, teks) -> teks)
    .count();

Di titik ini, operasi sudah tidak lagi stateless. count menyimpan penghitung per key pada state store lokal dan menulis perubahannya ke changelog topic, dan keduanya menjadi bahan utama topik State store. Kombinasi groupBy dan count semacam itulah yang menjadi jembatan dari primitif stateless menuju agregasi, sedangkan windowing melengkapi jalur tersebut dengan menambahkan batas waktu pada agregasi yang sama.

Simulasi pipeline pada daftar kejadian tetap#

Simulasi berikut memutar pipeline flatMap, filter, map, dan groupBy disertai count pada tiga baris catatan tetap, tanpa broker dan tanpa state store sungguhan. Setiap tahap diberi judul keluaran sendiri agar jumlah record masukan dan keluarannya dapat dibandingkan langsung. Daftar masukannya sengaja dibuat pendek agar hasil setiap tahap dapat dihitung ulang dengan tangan, misalnya jumlah kata yang dibuang filter atau kelompok yang terhitung lebih dari satu kali.

python primitif-streams.py 
# simulasi pipeline stateless: flatMap -> filter -> map -> groupBy + count

baris_kejadian = [
    "kafka menyimpan log kejadian",
    "log dibaca dari ujung ke ujung",
    "kejadian diputar ulang dari log",
]
PANJANG_MINIMUM = 4


def flat_map(baris):
    # satu record menjadi banyak record, seperti flatMapValues
    return baris.split(" ")


def filter_buang_pendek(kata):
    # record lolos bila predikatnya benar, seperti filter
    return len(kata) >= PANJANG_MINIMUM


def map_pasangan(kata):
    # nilai dipetakan menjadi pasangan (kata, 1), seperti map
    return (kata, 1)


kata_semua = [kata for baris in baris_kejadian for kata in flat_map(baris)]

print(f"flatMap: {len(baris_kejadian)} baris menjadi {len(kata_semua)} kata")
for urut, baris in enumerate(baris_kejadian):
    print(f"  baris {urut}: {baris} -> {flat_map(baris)}")

lolos = [kata for kata in kata_semua if filter_buang_pendek(kata)]
print(f"filter: kata kurang dari {PANJANG_MINIMUM} huruf dibuang")
print(f"  {len(kata_semua)} kata masuk, {len(lolos)} kata lolos")

pasangan = [map_pasangan(kata) for kata in lolos]
print("map: setiap kata menjadi pasangan (kata, 1)")
for kata, satu in pasangan:
    print(f"  ({kata}, {satu})")

penghitung = {}
for kata, satu in pasangan:
    penghitung[kata] = penghitung.get(kata, 0) + satu
print("groupBy + count: jumlahkan pasangan per key")
for kata, jumlah in sorted(penghitung.items()):
    print(f"  {kata} -> {jumlah}")
text Keluaran 
flatMap: 3 baris menjadi 15 kata
  baris 0: kafka menyimpan log kejadian -> ['kafka', 'menyimpan', 'log', 'kejadian']
  baris 1: log dibaca dari ujung ke ujung -> ['log', 'dibaca', 'dari', 'ujung', 'ke', 'ujung']
  baris 2: kejadian diputar ulang dari log -> ['kejadian', 'diputar', 'ulang', 'dari', 'log']
filter: kata kurang dari 4 huruf dibuang
  15 kata masuk, 11 kata lolos
map: setiap kata menjadi pasangan (kata, 1)
  (kafka, 1)
  (menyimpan, 1)
  (kejadian, 1)
  (dibaca, 1)
  (dari, 1)
  (ujung, 1)
  (ujung, 1)
  (kejadian, 1)
  (diputar, 1)
  (ulang, 1)
  (dari, 1)
groupBy + count: jumlahkan pasangan per key
  dari -> 2
  dibaca -> 1
  diputar -> 1
  kafka -> 1
  kejadian -> 2
  menyimpan -> 1
  ujung -> 2
  ulang -> 1

Tiga baris masukan menghasilkan 15 kata pada tahap flatMap, lalu filter menyisakan 11 kata setelah empat kata pendek dibuang. Tahap map menautkan angka satu pada tiap kata, dan groupBy disertai count menjumlahkan pasangan itu menjadi delapan kelompok, dengan dari, kejadian, dan ujung masing-masing terhitung dua kali. Delapan kelompok itu menampung seluruh 11 pasangan, sehingga tidak ada kata yang hilang di tengah pipeline. Keluaran terakhir berperilaku seperti KTable, yaitu satu nilai terkini per key, dan persis pada tahap inilah state store sungguhan mulai bekerja pada topologi nyata.

Kesalahan umum#

  1. Menggunakan map padahal key tidak berubah. Setiap perubahan key memicu topic repartisi internal yang menambah beban broker dan latensi, sedangkan mapValues sudah cukup untuk kebutuhan yang sama. Periksa kebutuhan key sebelum memilih operator pemetaannya.
  2. Menyusun join stream-stream tanpa window atau dengan window yang berlebihan. Tanpa window, API join stream-stream menolak dijalankan, sedangkan window yang terlalu luas menumpuk state dan memasangkan record yang sebenarnya tidak berkaitan. Tetapkan window berdasarkan jeda wajar antara kedua sumber kejadian.
  3. Menyimpan hasil transformasi sementara pada koleksi di luar topologi, misalnya variabel statis yang diisi dari dalam lambda. Keadaan semacam itu tidak ikut pulih saat task berpindah instance dan tidak dibagikan antar instance. Simpan keadaan yang memang dibutuhkan melalui operator DSL atau state store, bukan melalui koleksi aplikasi.
  4. Mengandalkan operasi stateless untuk logika yang sesungguhnya menyimpan keadaan, misalnya menghitung kemunculan dengan menaikkan variabel di dalam lambda filter. Variabel itu tidak dibagikan antar task dan hilang saat instance berganti, sehingga hasilnya tidak dapat diandalkan. Gunakan groupBy disertai agregasi untuk penghitungan semacam ini.

Ringkasan#

  • Operator stateless menentukan keluaran hanya dari record yang diproses, sehingga mudah diparalelkan dan pulih dengan membaca topic kembali.
  • filter menyaring dengan predikat, mapValues mengubah nilai saja, map mengubah key dan nilai, sedangkan flatMapValues melipatgandakan record dengan key yang sama.
  • Join stream-stream selalu berpasangan dengan window, dan groupBy menyusun record untuk agregasi sambil memicu repartisi.
  • count setelah groupBy adalah operasi stateful pertama yang menghasilkan KTable berbasis state store dan changelog topic.

Referensi#