Streams dan Connect

Alur minimum proyek sensor suhu sudah berjalan, yaitu layanan perekam menulis kejadian ke topic utama dan layanan perekap menghitung rangkuman dengan kode consumer yang ditulis sendiri. Topik ini menambahkan dua kemampuan berikutnya ke proyek, yaitu rangkuman real-time dengan Kafka Streams yang mengambil alih peran layanan perekap, dan integrasi Kafka Connect yang mengangkut rangkuman menuju basis data serta mengalirkan kembali kejadian tertahan dari gateway. Pembahasan ditutup dengan daftar periksa keberhasilan yang mengaitkan seluruh komponen ke angka kontraknya.

Rangkuman real-time dengan Kafka Streams#

Kode perekap manual menjalankan tiga pekerjaan yang berulang antar aplikasi, yaitu membaca kejadian, menyimpan kemajuan konsumsi, dan menulis hasil rangkuman ke topic. Kafka Streams, pustaka pemrosesan stream yang ditautkan ke aplikasi JVM, mengambil alih ketiganya sebagai satu deklarasi topologi, sehingga kode yang tersisa hanya logika perhitungan rangkumannya. Bagi broker, aplikasi Streams hanyalah consumer sekaligus producer biasa, sehingga tidak ada komponen baru yang dioperasikan di sisi cluster. Offset dan state aplikasi kini dikelola pustakanya, sedangkan kebijakan dedupe tetap menjadi bagian proyek.

Aplikasi Streams proyek ini membaca m8-suhu-kejadian dan menulis ke m8-suhu-rangkuman, persis mengikuti peta topic pada dokumen arsitektur. Identitas group aplikasi ditetapkan melalui application.id bernilai m8-perekap-streams mengikuti awalan proyek, dan paralelismenya mengikuti partisi input, yaitu topic dengan tiga partisi memberi paling banyak tiga task. Diagram berikut memperlihatkan alur lengkapnya sampai ke basis data rangkuman dan antrean tampilan.

Topologi window satu jam#

Dokumen arsitektur menetapkan rangkuman per jam per perangkat, dan topologi berikut mewujudkannya dengan window tumbling, yaitu window berukuran tetap yang tidak tumpang-tindih. Potongan Java di bawah adalah bagian dari proyek Maven atau Gradle dengan dependensi kafka-streams, sehingga keduanya berfungsi sebagai fragmen untuk dibaca dan tidak dieksekusi. Kelas pertama adalah akumulator state untuk satu window satu perangkat.

// tidak-dijalankan
record RangkumanSuhu(int jumlah, double total, double min, double max) {
    RangkumanSuhu tambah(double suhu) {
        return new RangkumanSuhu(jumlah + 1, total + suhu,
                Math.min(min, suhu), Math.max(max, suhu));
    }

    String toJson(String idPerangkat, Instant jamWindow) {
        return String.format(
            "{\"id_perangkat\": \"%s\", \"jam_window\": \"%s\", \"jumlah\": %d, "
                + "\"suhu_rata\": %.2f, \"suhu_min\": %.2f, \"suhu_max\": %.2f}",
            idPerangkat, jamWindow, jumlah, total / jumlah, min, max);
    }
}

RangkumanSuhu hanya menyimpan jumlah kejadian, total suhu, nilai minimum, dan nilai maksimum, karena angka rata-rata dan kedua ekstremnya cukup dihitung dari empat nilai itu. Operasi tambah bersifat komutatif, sehingga kejadian yang tiba dengan urutan apa pun menghasilkan akumulator yang sama. Metode toJson menuliskan payload rangkuman berisi id_perangkat, jam_window, dan empat angka hasilnya, sedangkan metode pembantu bacaSuhu yang dipanggil topologi mengambil nilai suhu_c dari field data pada envelope kejadian.

// tidak-dijalankan
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> kejadian = builder.stream("m8-suhu-kejadian");

KTable<Windowed<String>, RangkumanSuhu> agregat = kejadian
    .groupByKey()                                    // key tetap id_perangkat
    .windowedBy(TimeWindows.ofSize(Duration.ofHours(1))
                              .grace(Duration.ofMinutes(5)))
    .aggregate(
        () -> new RangkumanSuhu(0, 0.0, Double.MAX_VALUE, -Double.MAX_VALUE),
        (idPerangkat, envelope, r) -> r.tambah(bacaSuhu(envelope)),
        Materialized.as("rangkuman-jam")             // state lokal + changelog
    );

agregat.toStream()
    .map((window, r) -> KeyValue.pair(window.key(),
        r.toJson(window.key(), window.window().startTime())))
    .to("m8-suhu-rangkuman");

Properties properti = new Properties();
properti.put(StreamsConfig.APPLICATION_ID_CONFIG, "m8-perekap-streams");
properti.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
KafkaStreams aplikasi = new KafkaStreams(builder.build(), properti);
aplikasi.start();

Aliran record pada topologi mengikuti peta arsitektur. Metode builder.stream membentuk processor sumber dari m8-suhu-kejadian, groupByKey mempertahankan key id_perangkat, dan windowedBy beserta aggregate menyimpan akumulator window pada state store lokal bernama rangkuman-jam beserta changelog topic-nya. Hasil agregasi berupa tabel per window, sehingga toStream beserta map mengubahnya menjadi record baru berunci id_perangkat sebelum to menulisnya ke m8-suhu-rangkuman. Jeda lima menit pada grace menahan penutupan window bagi kejadian yang tiba terlambat, sedangkan kejadian yang melewatinya diabaikan. Batas window dihitung dari timestamp record, sedangkan waktu kejadian sesungguhnya berada pada field occurred_at, sehingga apabila jeda pengiriman mulai berarti, proyek memasang timestamp extractor, komponen yang menentukan waktu record dari field tersebut.

Kontrak pengiriman tetap berlaku pada alur baru ini. Kejadian ulang dari sesi producer yang sama sudah ditolak broker, sedangkan kejadian ulang lintas sesi ditahan oleh tahap dedupe event_id yang dipasang proyek sebagai satu processor sebelum aggregate, dan tahap itu tidak ada pada fragmen di atas. Nilai rangkuman per perangkat dan jam ditentukan penuh oleh isi kejadian, sehingga penulisan ulang menghasilkan nilai yang sama dan pemrosesan ulang setelah crash tidak mengubah data lama.

Simulasi agregasi window#

Broker dan pustaka Streams tidak diperlukan untuk memahami tahap agregasinya, sehingga simulasi berikut menjalankan topologi yang sama pada delapan kejadian tetap dengan Python murni. Tahap dedupe ikut disimulasikan karena ia bagian dari topologi proyek, dan window 08.00 serta 09.00 keduanya terisi supaya hasil per window dapat dibandingkan. Skrip dijalankan dari folder topik dengan perintah python3 -B simulasi-window-suhu.py.

# simulasi topologi rangkuman per jam: sumber -> dedupe -> aggregate window -> sink
# data uji tetap dan keluaran deterministik, tanpa broker, tanpa waktu sistem

kejadian = [
    ("ev-001", "sensor-01", "2026-11-07T08:05:00Z", 24.5),
    ("ev-002", "sensor-01", "2026-11-07T08:20:00Z", 24.7),
    ("ev-003", "sensor-02", "2026-11-07T08:15:00Z", 23.8),
    ("ev-004", "sensor-03", "2026-11-07T08:40:00Z", 22.9),
    ("ev-005", "sensor-03", "2026-11-07T08:45:00Z", 23.2),
    ("ev-006", "sensor-01", "2026-11-07T09:10:00Z", 25.1),
    ("ev-007", "sensor-02", "2026-11-07T09:05:00Z", 24.0),
    # kejadian ulang dengan event_id sama, meniru pengiriman ulang
    ("ev-002", "sensor-01", "2026-11-07T08:20:00Z", 24.7),
]


def jam_window(occurred_at):
    """Batas window satu jam ditentukan dari waktu kejadian."""
    return occurred_at[:13] + ":00:00Z"


print(f"sumber: {len(kejadian)} record dari m8-suhu-kejadian")

# processor dedupe: event_id yang sudah masuk perhitungan ditandai
seen = set()
lolos = []
dilewati = []
for eid, perangkat, waktu, suhu in kejadian:
    if eid in seen:
        dilewati.append((eid, perangkat))
        continue
    seen.add(eid)
    lolos.append((perangkat, waktu, suhu))
print(f"dedupe : {len(lolos)} kejadian unik, {len(dilewati)} kejadian ulang dilewati")

# processor aggregate: akumulator per (perangkat, window) dalam state lokal
akumulator = {}   # (perangkat, window) -> [jumlah, total, min, max]
for perangkat, waktu, suhu in lolos:
    kunci = (perangkat, jam_window(waktu))
    if kunci not in akumulator:
        akumulator[kunci] = [0, 0.0, suhu, suhu]
    a = akumulator[kunci]
    a[0] += 1
    a[1] += suhu
    a[2] = min(a[2], suhu)
    a[3] = max(a[3], suhu)

print("aggregate: rangkuman per perangkat per window satu jam")
window_sebelumnya = None
for (perangkat, window), (jumlah, total, minim, maksim) in sorted(
        akumulator.items(), key=lambda kv: (kv[0][1], kv[0][0])):
    if window != window_sebelumnya:
        print(f"  window {window}")
        window_sebelumnya = window
    rata = total / jumlah
    print(f"    {perangkat} jumlah={jumlah} suhu_rata={rata:.2f} suhu_min={minim} suhu_max={maksim}")

# processor sink: setiap rangkuman ditulis ke m8-suhu-rangkuman berunci perangkat
print(f"sink   : {len(akumulator)} rangkuman ditulis ke m8-suhu-rangkuman (key id_perangkat)")

sumber: 8 record dari m8-suhu-kejadian
dedupe : 7 kejadian unik, 1 kejadian ulang dilewati
aggregate: rangkuman per perangkat per window satu jam
  window 2026-11-07T08:00:00Z
    sensor-01 jumlah=2 suhu_rata=24.60 suhu_min=24.5 suhu_max=24.7
    sensor-02 jumlah=1 suhu_rata=23.80 suhu_min=23.8 suhu_max=23.8
    sensor-03 jumlah=2 suhu_rata=23.05 suhu_min=22.9 suhu_max=23.2
  window 2026-11-07T09:00:00Z
    sensor-01 jumlah=1 suhu_rata=25.10 suhu_min=25.1 suhu_max=25.1
    sensor-02 jumlah=1 suhu_rata=24.00 suhu_min=24.0 suhu_max=24.0
sink   : 5 rangkuman ditulis ke m8-suhu-rangkuman (key id_perangkat)

Delapan record masuk ke tahap sumber, satu di antaranya kejadian ulang dengan event_id sama sehingga tahap dedupe menyisakan tujuh kejadian. Window 08.00 berisi lima kejadian dari tiga perangkat, dengan sensor-01 mencatat rata-rata 24.60 dari dua pembacaan, sedangkan window 09.00 berisi dua kejadian dari dua perangkat. Lima rangkuman berakhir di sink dengan key id_perangkat, dan setiap angka pada keluaran dapat dihitung ulang pembaca dari data ujinya. Pada aplikasi sungguhan, akumulator diperbarui setiap kali kejadian masuk dan hasilnya dikirim ke topic tanpa menunggu window ditutup, berbeda dengan simulasi yang mencetak seluruh hasilnya sekaligus.

Integrasi Connect ke basis data rangkuman#

Sejarah rangkuman yang perlu ditinjau lintas hari tidak cukup disajikan dari topic saja, sehingga proyek menyalin rangkuman ke basis data relasional yang dapat dikueri. Kafka Connect memindahkan data itu tanpa kode consumer baru, yaitu melalui sink JDBC yang berjalan di worker Connect sebagai plugin, sedangkan bagi broker task connector hanyalah consumer topic rangkuman. Konfigurasi connector disiapkan sebagai berkas JSON yang didaftarkan ke REST API worker.

{
  "name": "sink-jdbc-suhu",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "topics": "m8-suhu-rangkuman",
    "connection.url": "jdbc:postgresql://localhost:5432/rangkuman_suhu",
    "insert.mode": "upsert",
    "pk.mode": "record_value",
    "pk.fields": "data.id_perangkat,data.jam_window",
    "transforms": "flatten",
    "transforms.flatten.type": "org.apache.kafka.connect.transforms.Flatten$Value",
    "auto.create": "true",
    "auto.evolve": "true"
  }
}

Nilai insert.mode=upsert membuat task menulis rangkuman dengan pernyataan penyisipan yang berubah menjadi pembaruan ketika barisnya sudah ada. Primary key-nya diambil dari nilai record melalui pk.mode=record_value dengan pk.fields yang menyebut data.id_perangkat dan data.jam_window, sedangkan Flatten$Value, sebuah single message transform (SMT) bawaan Connect, memangkatkan envelope sehingga field di dalam data menjadi field tingkat atas bernama data.id_perangkat dan data.jam_window. Kombinasi tersebut menghasilkan satu baris per perangkat per jam, dan rangkuman yang terkirim ulang menimpa baris yang sama dengan nilai yang identik, sehingga kontrak alur penulisan rangkuman tetap terpenuhi. Kunci auto.create beserta auto.evolve menyusun tabel dari skema record pertama.

Basis data rangkuman melengkapi, bukan menggantikan, jalur lama. Antrean tampilan tetap membaca m8-suhu-rangkuman untuk nilai terkini, sedangkan tabel hasil sink menyimpan sejarah yang dikueri untuk laporan. Dengan begitu, keputusan retensi tiga puluh hari pada topic rangkuman tetap menyokong antrean tampilan sebagai pembaca alirannya.

Impor kejadian tertahan dengan source file#

Gateway menyimpan pembacaan ke berkas lokal ketika jalur menuju broker terputus, dan berkas itu dialirkan kembali begitu jalurnya pulih. Distribusi Kafka membawa FileStreamSource yang membaca berkas baris demi baris dan menulis setiap baris sebagai nilai record ke topic tujuan, sehingga jalur pemulihan ini cukup disiapkan dengan konfigurasi dan tanpa kode tambahan. Konfigurasi berikut mengalirkan berkas kejadian tertahan ke topic utama.

{
  "name": "source-file-suhu",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
    "tasks.max": "1",
    "file": "/var/log/gateway/suhu-tertahan.log",
    "topic": "m8-suhu-kejadian"
  }
}

Dua sifat connector ini menentukan cara penggunaannya. Setiap baris berkas harus sudah berbentuk envelope lengkap dengan event_id, karena gateway yang menuliskan kejadian gagal kirim sudah memuat semuanya, dan dedupe event_id di perekap menahan baris yang terulang karena proses impor dijalankan dua kali. Baris juga ditulis tanpa key, sehingga partisi tujuannya dipilih tanpa pemetaan id_perangkat, tetapi perhitungan rangkuman bersifat komutatif dan batas window mengikuti occurred_at, sehingga hasilnya tidak berubah.

Daftar periksa keberhasilan proyek#

Komponen proyek bertambah, tetapi ukuran keberhasilannya tidak berubah, yaitu angka kontrak yang ditetapkan saat pemilihan proyek. Setiap butir kontrak dikaitkan dengan cara pemeriksaan konkret dan komponen yang menyumbangnya, sehingga pemeriksaan tidak lagi berupa penilaian, melainkan kegiatan yang dapat dijadwalkan. Butir yang tidak dapat dipetakan ke satu cara pemeriksaan menandai angka kontrak yang perlu dirumuskan ulang pada peninjauan berikutnya. Pemeriksaan dijalankan dengan urutan yang sama pada setiap siklus, sehingga hasilnya dapat dibandingkan antar siklus.

Proyek dinyatakan selesai ketika seluruh butir dapat diperiksa tanpa pengecualian, dan daftar yang sama menjadi pegangan saat komponen baru ditambahkan. Setiap baris daftar periksa sebaiknya dijadikan skrip pemeriksaan kecil yang dijalankan berkala, karena pemeriksaan manual cepat tergeser oleh pekerjaan operasional. Skrip lag perekap pada topik Implementasi producer dan consumer menyediakan dasar untuk butir pertama. Perubahan arsitektur berikutnya wajib memperbarui daftar ini bersama dokumen arsitekturnya, sehingga angka yang diperjanjikan tidak tertinggal dari sistem yang benar-benar berjalan. Tinjauan berkala juga menandai baris yang perlu disesuaikan ketika lingkup proyek berubah.

Ringkasan#

• Aplikasi Streams dengan identitas m8-perekap-streams mengambil alih peran perekap, dan topologi window satu jam menghitung rangkuman per perangkat dari m8-suhu-kejadian menuju m8-suhu-rangkuman.
• Akumulator window bersifat komutatif dan rangkumannya ditulis ulang dengan nilai sama, sehingga pemrosesan ulang tidak mengubah data, sedangkan dedupe event_id tetap berlaku di depan agregasi.
• Sink JDBC memindahkan rangkuman ke basis data dengan upsert per perangkat per jam, dan FileStreamSource mengalirkan kembali kejadian tertahan dengan batasan yang dijelaskan.
• Daftar periksa keberhasilan mengaitkan setiap angka kontrak dengan cara pemeriksaan dan komponen penyumbangnya.

Referensi#

• Kafka Streams — Apache Software Foundation — gambaran pustaka Streams beserta model aplikasi dan paralelismenya.
• Streams DSL — Apache Software Foundation — operator groupByKey, windowedBy, dan aggregate yang menyusun topologi proyek.
• Kafka Connect User Guide — Apache Software Foundation — aturan worker, plugin, transformasi, dan pendaftaran connector yang digunakan integrasi.
• JDBC Sink Connector for Confluent Platform — Confluent Documentation — semantik upsert, mode primary key, dan auto-creation pada sink.
• Apache Kafka documentation — Apache Software Foundation — dokumentasi utama yang mencakup FileStream connector dan transformasi bawaan Connect.