Belajar Apache Flink - State Management & Checkpointing

#Pendahuluan

Episode 5 mengenalkan windowing yang menyimpan agregasi sementara. Episode 6 ini mengangkat hal itu ke permukaan: state adalah salah satu pembeda terbesar Flink dibanding engine streaming lain. Hampir semua analitik yang berguna — total per pengguna, deteksi pola, rekomendasi — membutuhkan kemampuan mengingat apa yang sudah terjadi.

Kita akan membedah jenis-jenis state, cara menyimpan dan mengaksesnya lewat ValueState, ListState, MapState, dan AggregatingState, lalu mengonfigurasi checkpointing untuk menjamin exactly-once. Terakhir, restart strategies menjadikan job tahan banting: jika sebuah TaskManager mati, job pulih dengan sendirinya.

#Stateful Operators dan Keyed State

#Kapan Operator Menyimpan State

Operator dikatakan stateful jika ia menyimpan data di antara elemen yang masuk. Contoh paling sederhana: operator sum di dalam window, atau operator yang menghitung total berjalan per pengguna. Tanpa state, keduanya mustahil dilakukan — setiap event diproses tanpa ingatan apa pun.

#Keyed State vs Operator State

Flink membedakan dua keluarga state:

• Keyed state: state yang di-scope ke satu key, dipakai setelah operator keyBy. Contohnya total transaksi per userId.
• Operator state: state yang melekat pada satu subtask operator, tidak peduli key apa pun. Contohnya list buffer yang dipakai source saat mengambil posisi konsumsi.

keyed state   → per key, diakses dari keyed context
operator state → per subtask, dibagikan antar operator terkait

#Tipe State di Flink

#ValueState

ValueState menyimpan satu nilai per key. Cocok untuk melacak nilai terbaru atau akumulator sederhana:

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
 
ValueStateDescriptor<Long> desc =
    new ValueStateDescriptor<Long>("total", Long.class);
 
ValueState<Long> total = getRuntimeContext().getState(desc);
Long lama = total.value();
total.update((lama == null ? 0L : lama) + event.getAmount());

total.value() mengambil nilai saat ini, dan total.update menimpanya. Pola baca-tulis-update seperti ini adalah idiom utama keyed state.

#ListState dan MapState

ListState menyimpan kumpulan elemen, MapState menyimpan pasangan key-value. Keduanya dipakai saat state harus menampung lebih dari satu entitas:

ListStateDescriptor<String> riwayatDesc =
    new ListStateDescriptor<String>("riwayat", String.class);
 
ListState<String> riwayat = getRuntimeContext().getListState(riwayatDesc);
riwayat.add(event.getId());

MapStateDescriptor<String, Double> hargaDesc =
    new MapStateDescriptor<String, Double>(
        "harga", String.class, Double.class);
 
MapState<String, Double> harga = getRuntimeContext().getMapState(hargaDesc);
harga.put("apel", 15000.0);
double apel = harga.get("apel");

getListState dan getMapState mengikuti pola yang sama: deklarasikan descriptor, lalu dapatkan handle dari runtime context. Pilih ListState saat urutan penting, MapState saat pencarian cepat berdasarkan sub-key dibutuhkan.

#AggregatingState

AggregatingState menggabungkan seluruh nilai menjadi satu hasil agregat tanpa menyimpan elemen aslinya — efisien untuk sum, average, atau hitungan:

AggregatingStateDescriptor<Order, Long, Long> aggDesc =
    new AggregatingStateDescriptor<Order, Long, Long>(
        "total-agg",
        new AggregateFunction<Order, Long, Long>() {
            public Long createAccumulator() { return 0L; }
            public Long add(Order value, Long acc) { return acc + value.getAmount(); }
            public Long getResult(Long acc) { return acc; }
            public Long merge(Long a, Long b) { return a + b; }
        },
        Long.class);
 
AggregatingState<Order, Long> agg =
    getRuntimeContext().getAggregatingState(aggDesc);

AggregateFunction di dalam descriptor mendefinisikan cara inisialisasi, menambah, mengambil hasil, dan menggabungkan accumulator — fungsi inilah yang menentukan logika agregasi.

#Konfigurasi Checkpointing

#Checkpoint Otomatis

Checkpoint adalah snapshot periodik dari seluruh state plus posisi konsumsi source. Agar otomatis, aktifkan interval di config.yaml:

execution.checkpointing.interval: 5min
execution.checkpointing.mode: exactly-once
execution.checkpointing.min-pause: 1min
execution.checkpointing.timeout: 10min
execution.checkpointing.tolerable-failed-checkpoints: 2

execution.checkpointing.mode memilih antara exactly-once dan at-least-once. min-pause menjamin checkpoint tidak saling tumpuk, dan tolerable-failed-checkpoints menoleransi kegagalan kecil sebelum job dianggap gagal.

#Jaminan Konsistensi

Checkpointing Flink berbasis distributed snapshots (algoritma Chandy-Lamport yang diadaptasi). Saat proses mati, Flink me-restore state ke checkpoint terakhir dan replay event dari posisi source yang tersimpan. Karena event diproses ulang, jaminan exactly-once tercapai asalkan sink juga mendukung dua fase (seperti Kafka sink) atau idempoten.

#Memilih State Backend

State disimpan di state backend:

state.backend.type: rocksdb
state.backend.incremental: true
state.backend.local-recovery: true

Backend hashmap menyimpan state di heap memory — cepat namun terbatas. Backend rocksdb menyimpan di disk lokal dengan cache di memory — mampu menampung state raksasa. Incremental checkpoint membuat snapshot lebih ringan karena hanya mengirim perubahan sejak checkpoint terakhir.

#Restart Strategies dan Fault Tolerance

#Strategi Restart

Restart strategy menentukan bagaimana job pulih setelah gagal. Di config.yaml:

restart-strategy.type: fixed-delay
restart-strategy.fixed-delay.attempts: 3
restart-strategy.fixed-delay.delay: 10s

Ada juga mode exponential-delay yang menunda restart semakin lama setiap percobaan — cocok untuk produksi agar sistem hilir tidak kebanjiran retry. Kombinasi restart strategy dan checkpointing inilah yang membuat Flink bersifat self-healing.

#Memantau Checkpoint

Pantau status checkpoint dari REST API atau dashboard:

curl -s http://localhost:8081/jobs/overview
curl -s http://localhost:8081/jobs | python3 -m json.tool

Perintah curl -s http://localhost:8081/jobs menampilkan daftar job dalam bentuk JSON. Di dashboard, tab Checkpoints menampilkan riwayat snapshot: ukuran state, durasi, dan status SUCCESS atau FAILED.

#Penutup

Episode 6 membekali kalian dengan fondasi keandalan: memahami keyed state dan operator state, menguasai ValueState, ListState, MapState, dan AggregatingState, mengonfigurasi checkpointing dengan mode exactly-once, memilih state backend yang tepat, serta menyusun restart strategy agar job pulih sendiri.

Inti yang harus dibawa pulang:

• Keyed state melekat pada key setelah keyBy; operator state melekat pada subtask.
• ValueState, ListState, MapState, dan AggregatingState dipilih sesuai bentuk data state.
• Checkpointing otomatis menjamin exactly-once dengan snapshot state dan posisi source.
• Backend hashmap memakai heap memory; backend rocksdb memakai disk dan cocok untuk state besar.
• Restart strategy menentukan cara job pulih setelah kegagalan, melengkapi checkpointing.

Di episode 7 selanjutnya kita akan membahas error handling dan debugging stream jobs — menangani exception di operator dan source, memantau status job dan log, memakai savepoints untuk recovery, serta debugging dengan local cluster dan flink run -d. Keterampilan ini yang membedakan engineer yang menulis kode dari yang mengoperasikan sistem.