Ilustrasi dihasilkan AI untuk artikel ini. Kebanyakan sistem analitik dibangun dengan asumsi bahwa data diam: dikumpulkan dulu, lal...
Kebanyakan sistem analitik dibangun dengan asumsi bahwa data diam: dikumpulkan dulu, lalu diproses dalam batch. Asumsi itu runtuh ketika keputusan bisnis harus diambil dalam hitungan detik, misalnya mendeteksi transaksi mencurigakan saat transaksi itu terjadi, atau menghitung agregat perangkat IoT ketika gangguannya masih berlangsung. Di titik inilah Apache Flink menjadi salah satu rujukan utama arsitektur data modern: ia dirancang sejak awal sebagai mesin stateful stream processing, bukan mesin batch yang diberi tambahan mode streaming.
Artikel ini membahas tiga hal yang paling sering disalahpahami ketika tim mulai memakai Flink: bagaimana state dikelola, bagaimana checkpoint sebenarnya bekerja, dan sejauh mana klaim exactly-once boleh dipercaya. Pembahasan disertai contoh konfigurasi yang bisa dijalankan di laptop, serta catatan jujur mengenai batas pengujiannya.
Mengapa stateful stream processing mengubah cara memandang data
Pada pemrosesan tanpa state, setiap record diproses secara independen. Model itu mudah diskalakan dan mudah diuji, tetapi tidak mampu menjawab pertanyaan yang membutuhkan memori: berapa transaksi per kartu dalam lima menit terakhir, apakah suhu sensor naik lebih dari sepuluh derajat dibanding rata-rata satu jam sebelumnya, atau apakah sesi pengguna ini sudah melewati ambang kegagalan login. Semua pertanyaan tersebut membutuhkan state yang hidup lebih lama daripada satu record dan harus konsisten meskipun terjadi kegagalan.
Flink menempatkan state sebagai warga kelas satu. Operator dapat menyimpan nilai terhitung per key, memperbaruinya pada setiap event, dan menjalankan timer yang dipicu oleh waktu event. Karena state itu tersimpan dan berversi melalui checkpoint, Flink dapat memulihkan aplikasi ke titik konsisten setelah crash, penskalaan ulang, atau upgrade kode. Konsekuensinya, desain aplikasi streaming bukan lagi soal "bagaimana memproses satu pesan", melainkan soal "state apa yang perlu dipertahankan, berapa lama, dan seberapa besar".
Arsitektur runtime: JobManager, TaskManager, dan distribusi state
Sebuah job Flink dijalankan oleh dua jenis proses. JobManager bertanggung jawab atas koordinasi: menerjemahkan job graph, menjadwalkan task, mengoordinasikan checkpoint, dan memulihkan kegagalan. TaskManager adalah worker yang mengeksekusi operator dan menyimpan state aktual di memori, disk lokal, atau state backend terdistribusi.
Hampir semua masalah operasional Flink muncul dari ketidakseimbangan antar peran itu: JobManager yang kewalahan karena terlalu banyak checkpoint kecil, TaskManager yang kehabisan heap karena state tidak terkontrol, atau slot yang tidak cukup sehingga job terjebak pada status CREATED.
Slot, paralelisme, dan key groups
Paralelisme menentukan berapa banyak salinan paralel sebuah operator. Setiap salinan menempati satu task slot pada TaskManager. Pada operator yang bekerja per key, Flink membagi ruang key menjadi key groups, dan setiap subtask bertanggung jawab atas sebagian key group. Mekanisme inilah yang memungkinkan rescaling: ketika paralelisme berubah, key group dipindahkan antar subtask tanpa kehilangan konsistensi, selama state dapat di-redistribusi dari checkpoint atau savepoint.
Praktik yang lazim adalah menetapkan maxParallelism secara eksplisit di awal, misalnya 128 atau 256, karena nilai ini menjadi batas atas paralelisme di masa depan. Mengubahnya setelah job berjalan berarti memulai dari state kosong.
Keyed state dan operator state
Flink membedakan dua keluarga state. Keyed state terikat pada key dan hanya dapat diakses di dalam konteks keyed stream; bentuknya mencakup ValueState, ListState, MapState, serta state agregasi seperti ReducingState. Operator state terikat pada instance operator, bukan pada key, dan umumnya dipakai untuk source yang perlu mengingat posisi baca, misalnya offset Kafka per partition.
Perbedaan ini bukan sekadar terminologi. Keyed state bisa di-redistribusi otomatis saat rescaling, sedangkan operator state memerlukan logika pembagian ulang, misalnya round-robin untuk ListState atau union untuk skema tertentu. Kesalahan memilih jenis state termasuk penyebab paling umum checkpoint gagal setelah perubahan paralelisme.
Menjalankan session cluster lokal dengan Docker Compose
Untuk eksplorasi, session cluster adalah pilihan paling ringan: satu JobManager, satu TaskManager, dan job dikirim ke cluster yang sudah hidup. Konfigurasi berikut adalah contoh laboratorium lokal, bukan konfigurasi produksi — tidak ada high availability, tidak ada penyimpanan objek, dan semua port terekspos ke host.
# docker-compose.yml — Flink 1.19 session cluster (contoh lab lokal)
services:
jobmanager:
image: flink:1.19-scala_2.12-java17
command: jobmanager
ports:
- "8081:8081" # Web UI
- "6123:6123" # RPC internal
environment:
- |
FLINK_PROPERTIES=
jobmanager.rpc.address: jobmanager
jobmanager.memory.process.size: 1600m
state.backend.type: hashmap
state.checkpoints.dir: file:///opt/flink/checkpoints
execution.checkpointing.interval: 30s
execution.checkpointing.mode: EXACTLY_ONCE
volumes:
- flink-checkpoints:/opt/flink/checkpoints
taskmanager:
image: flink:1.19-scala_2.12-java17
command: taskmanager
depends_on:
- jobmanager
environment:
- |
FLINK_PROPERTIES=
jobmanager.rpc.address: jobmanager
taskmanager.numberOfTaskSlots: 2
taskmanager.memory.process.size: 2048m
state.backend.type: hashmap
state.checkpoints.dir: file:///opt/flink/checkpoints
volumes:
flink-checkpoints:Setelah docker compose up -d, UI tersedia di http://localhost:8081. Perhatikan bahwa state.checkpoints.dir harus sama di JobManager dan TaskManager, dan pada klaster nyata sebaiknya menunjuk ke sistem penyimpanan terdistribusi seperti HDFS atau object storage — dokumentasi Checkpoints merinci persyaratan ini. Direktori lokal hanya memadai untuk eksperimen satu mesin, karena checkpoint hilang bersama container.
Job pertama: keyed state dan checkpoint dalam PyFlink
Contoh berikut membaca stream JSON dari socket lokal, melakukan keyBy pada card_id, lalu menghitung jumlah dan total nilai transaksi per kartu menggunakan ValueState. Ini pola yang sama dengan kasus deteksi fraud sederhana, tanpa klaim performa apa pun.
# flink_job.py — contoh lab: keyed state + checkpoint
from pyflink.common import Types, WatermarkStrategy, Duration
from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode
from pyflink.datastream.functions import KeyedProcessFunction, RuntimeContext
from pyflink.datastream.state import ValueStateDescriptor
class FraudWindow(KeyedProcessFunction):
def open(self, ctx: RuntimeContext):
self.count = ctx.get_state(ValueStateDescriptor("count", Types.INT()))
self.total = ctx.get_state(ValueStateDescriptor("total", Types.DOUBLE()))
def process_element(self, value, ctx):
c = self.count.value() or 0
t = self.total.value() or 0.0
c += 1
t += float(value["amount"])
self.count.update(c)
self.total.update(t)
yield {"card_id": value["card_id"], "n": c, "total": round(t, 2)}
env = StreamExecutionEnvironment.get_execution_environment()
env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
env.set_parallelism(2)
# Checkpoint tiap 30 detik, minimal pause 10 detik, toleransi 2 kegagalan
env.enable_checkpointing(30_000, 10_000)
env.get_checkpoint_config().set_tolerable_checkpoint_failure_number(2)
stream = (env.socket_text_stream("host.docker.internal", 9999)
.map(lambda s: __import__("json").loads(s))
.assign_timestamps_and_watermarks(
WatermarkStrategy.for_bounded_out_of_orderness(Duration.of_seconds(5)))
.key_by(lambda r: r["card_id"], key_type=Types.STRING())
.process(FraudWindow(), output_type=Types.PICKLED_BYTE_ARRAY()))
stream.print()
env.execute("flink-keyed-state-lab")Perhatikan dua baris penting: enable_checkpointing mengaktifkan checkpoint berkala, dan key_by memastikan state terpartisi per kartu. Tanpa keyBy, get_state akan gagal karena keyed state hanya tersedia pada konteks keyed. Selain itu, deklarasi tipe state secara eksplisit lebih aman daripada mengandalkan inferensi serialisasi, terutama ketika skema record berubah antar versi job.
Anatomi checkpoint
Checkpoint Flink bersifat asinkron dan terdistribusi. JobManager menyuntikkan barrier ke dalam stream; setiap operator, ketika menerima barrier, menyelesaikan pemrosesan record sebelum barrier, lalu menulis snapshot state-nya ke storage tanpa menghentikan aliran data. Setelah semua operator mengonfirmasi, JobManager menandai checkpoint sebagai selesai dan menyimpan metadata di _metadata.
Karena barrier harus melewati seluruh operator dalam waktu yang wajar, checkpoint dapat gagal atau timeout ketika ada backpressure berat atau satu subtask tertinggal. Itulah sebabnya pemantauan durasi checkpoint sama pentingnya dengan pemantauan latency job. Checkpoint yang terus-menerus gagal tidak menghilangkan jaminan konsistensi, tetapi membuat waktu pemulihan menjadi tidak dapat diprediksi.
Konfigurasi checkpoint storage dan restart strategy
Pada klaster yang lebih serius, checkpoint sebaiknya diarahkan ke sistem penyimpanan yang tahan kegagalan node, dan strategi restart ditetapkan secara eksplisit agar kegagalan tidak berubah menjadi loop tak terkendali. Potongan konfigurasi berikut menggunakan flink-conf.yaml; nilai-nilainya adalah titik awal yang wajar untuk lab, bukan hasil tuning produksi.
# flink-conf.yaml — checkpoint + restart strategy (contoh lab)
state.backend.type: hashmap
state.backend.incremental: true
state.checkpoints.dir: s3://flink-state/checkpoints
state.savepoints.dir: s3://flink-state/savepoints
execution.checkpointing.mode: EXACTLY_ONCE
execution.checkpointing.interval: 60s
execution.checkpointing.min-pause: 30s
execution.checkpointing.timeout: 10min
execution.checkpointing.max-concurrent-checkpoints: 1
execution.checkpointing.tolerable-failed-checkpoints: 3
execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
restart-strategy.type: exponential-delay
restart-strategy.exponential-delay.initial-backoff: 10s
restart-strategy.exponential-delay.max-backoff: 5min
restart-strategy.exponential-delay.backoff-multiplier: 1.5max-concurrent-checkpoints: 1 membatasi penumpukan snapshot dan biasanya cukup untuk workload yang tidak sangat besar. externalized-checkpoint-retention memastikan checkpoint tetap tersimpan ketika job dibatalkan, sehingga job dapat dilanjutkan dari titik yang sama. Strategi exponential-delay lebih disukai daripada fixed-delay untuk kegagalan yang berulang, karena memberi ruang bagi sistem eksternal pulih sebelum job dicoba lagi.
Catatan penting soal exactly-once. Flink memberikan jaminan state yang konsisten setelah pemulihan, tetapi exactly-once end-to-end hanya tercapai jika sumber dan tujuan sama-sama mendukungnya. Source seperti Kafka dengan offset yang dapat diputar ulang dan sink transaksional atau idempoten adalah syaratnya. Tanpa itu, Flink dapat mengirim ulang record setelah restart, dan sistem hilir harus menoleransi duplikasi. Klaim exactly-once yang tidak menyebut kondisi ini adalah klaim yang menyesatkan.
Exactly-once: apa yang benar-benar dijamin Flink
Flink menerapkan mekanisme yang secara longgar menyerupai two-phase commit untuk sink yang mendukung transaksi. Pada checkpoint, sink menandai transaksi yang sedang berjalan sebagai bagian dari snapshot. Ketika checkpoint selesai, transaksi tersebut dikomit. Jika terjadi kegagalan sebelum checkpoint selesai, transaksi dibatalkan dan record diputar ulang dari checkpoint terakhir. Hasilnya, record tidak hilang, dan tidak muncul dua kali di sistem hilir yang mendukung transaksi.
Yang sering terlewat: jaminan ini berlaku pada tepi sistem yang mendukung protokol tersebut. Untuk sink tanpa transaksi, Flink hanya bisa menjamin at-least-once. Sebagian implementasi menambahkan strategi idempoten — misalnya upsert berbasis primary key atau penulisan ke objek dengan nama deterministik — sehingga pengulangan tidak mengubah hasil akhir. Strategi itu sah, tetapi merupakan upaya di sisi aplikasi, bukan sihir dari Flink.
Batas jaminan: source dan sink adalah penentu
Source yang tidak dapat diputar ulang, seperti socket atau sumber push tanpa buffer, tidak dapat menjamin pemulihan yang tepat. Job yang membaca dari socket — termasuk contoh di atas — akan kehilangan data yang dikirim saat job mati, dan ini normal untuk eksperimen. Untuk sistem nyata, gunakan source dengan offset yang tercatat, misalnya Kafka atau file dengan posisi baca yang tersimpan di operator state. Di sisi sink, periksa dokumentasi konektor: dukungan transaksi berbeda antar versi dan antar sistem.
Verifikasi lewat UI dan CLI
Verifikasi tidak boleh berhenti pada "job berjalan". Yang perlu diperiksa adalah apakah checkpoint benar-benar selesai, berapa ukurannya, dan apakah state bertumbuh tanpa batas. UI Flink menyediakan tab Checkpoints per job; CLI memberi jalan cepat untuk otomatisasi.
# Verifikasi cepat dari host (CLI Flink di dalam container jobmanager)
docker compose exec jobmanager flink list -r
docker compose exec jobmanager flink list -a
# Memicu savepoint, lalu periksa isi direktori checkpoint
docker compose exec jobmanager flink savepoint <job-id> file:///opt/flink/checkpoints
ls -lh ./flink-checkpoints/<job-id>/chk-*
# Ringkasan checkpoint terakhir via REST API
curl -s http://localhost:8081/jobs/<job-id>/checkpoints | head -c 800Pada respons REST, perhatikan field status (harus COMPLETED), end_to_end_duration (lama checkpoint dari suntikan barrier hingga selesai), dan state_size. Kenaikan state_size yang terus-menerus tanpa perubahan beban menandakan adanya key yang tidak pernah kedaluwarsa — biasanya karena state TTL tidak dikonfigurasi. Detail format dan opsi lanjutan tersedia di dokumentasi Flink, sementara pola operator stateful dijelaskan pada bagian DataStream API overview.
Perlu dicatat: seluruh contoh pada artikel ini disusun sebagai skenario laboratorium lokal. Konfigurasi belum diuji end-to-end pada klaster produksi, dan angka konkret mengenai throughput maupun latency sengaja tidak dicantumkan karena sangat bergantung pada perangkat keras, pola key, dan ukuran record. Pembaca yang membutuhkan angka sebaiknya melakukan benchmark pada topologinya sendiri.
Pola produksi dan kesalahan yang sering terulang
Beberapa pola berulang muncul di hampir setiap implementasi Flink yang saya temui. Pertama, ukuran state dibiarkan tumbuh karena TTL tidak ditetapkan; solusinya bukan menambah memori, melainkan mendefinisikan masa hidup state sesuai kebutuhan bisnis. Kedua, checkpoint dijadwalkan terlalu sering sehingga metadata menumpuk dan JobManager menjadi titik tunggal yang lambat. Ketiga, paralelisme dinaikkan tanpa memeriksa jumlah partition source, sehingga sebagian subtask menganggur sementara yang lain kewalahan.
Backpressure, ukuran state, dan checkpoint yang menumpuk
Backpressure bukan gejala yang berdiri sendiri. Pada UI, indikator backpressure yang tinggi di satu operator biasanya berarti operator hilir lebih lambat, misalnya karena panggilan jaringan sink atau karena key skew. Key skew adalah kasus yang paling sering luput: distribusi key yang tidak merata membuat satu subtask memikul sebagian besar state dan trafik, sehingga checkpoint lambat karena satu subtask tertinggal. Memantau distribusi key dan menambah lapisan pre-agregasi sering kali lebih efektif daripada menambah TaskManager.
Savepoint sebagai kontrak operasional
Savepoint memungkinkan penghentian job secara terkendali untuk upgrade kode atau perubahan topologi. Kebiasaan yang sehat adalah mengambil savepoint sebelum setiap perubahan yang menyentuh state schema, menamainya secara eksplisit, dan menyimpan metadata versi job bersamanya. Tanpa disiplin itu, rollback menjadi mahal karena satu-satunya pilihan adalah memproses ulang data dari awal.
Dengan tiga pilar ini — state yang dikelola sadar, checkpoint yang dipantau sebagai indikator kesehatan, dan pemahaman jujur atas batas exactly-once — Flink bukan sekadar alat untuk "streaming cepat", melainkan fondasi bagi pipeline yang dapat dipulihkan dan diaudit. Selebihnya adalah pekerjaan disiplin operasional: mengukur, mencatat, dan tidak mengklaim jaminan yang tidak didukung oleh sistem di kedua ujung aliran.
COMMENTS