Belajar Apache Spark - Checkpointing & Fault Tolerance

#Pendahuluan

Aplikasi streaming di produksi berjalan berhari-hari atau bahkan berbulan-bulan tanpa henti. Selama itu, berbagai hal bisa terjadi: node mati, jaringan putus, Kafka tidak responsif. Episode 16 ini membahas bagaimana Spark tetap andal di tengah semua itu — lewat checkpointing dan fault tolerance.

Tanpa mekanisme pemulihan, satu kegagalan kecil bisa memaksa job streaming restart dari awal, kehilangan data yang sudah diproses atau menghasilkan duplikat. Memahami cara Spark menyimpan progress dan state adalah kunci untuk membangun pipeline streaming yang bisa dipercaya.

Episode ini membahas empat hal: checkpointing di Structured Streaming, perilaku fault tolerance dan recovery, state management untuk streaming jobs, serta best practices untuk aplikasi streaming jangka panjang.

#Checkpointing di Structured Streaming

#Apa yang Disimpan di Checkpoint

Checkpoint adalah lokasi di mana Spark menyimpan metadata pemrosesan secara berkala. Isinya mencakup:

• Offset log: posisi terakhir data yang dibaca dari setiap sumber — ini yang mencegah kehilangan data saat restart.
• State metadata: definisi query dan konfigurasi yang sedang berjalan.
• State data: untuk operator ber-state, snapshot state disimpan juga.

Menentukan checkpoint location

query = df.writeStream \
    .format("parquet") \
    .option("path", "data/hasil") \
    .option("checkpointLocation", "data/checkpoint") \
    .outputMode("append") \
    .start()

checkpointLocation harus disimpan di penyimpanan yang bertahan lama (HDFS, S3, atau volume Kubernetes yang persisten) — bukan di local disk worker yang bisa hilang. Dari metadata inilah Spark tahu harus melanjutkan dari mana.

#Menempatkan Checkpoint dengan Benar

Aturan penting: checkpoint location tidak boleh dipakai bersama oleh dua query yang berbeda, dan tidak boleh dipakai ulang oleh query yang strukturnya berubah drastis. Jika logika berubah signifikan, gunakan lokasi checkpoint baru.

#Fault Tolerance dan Recovery

#Dari Gagal ke Recovery

Saat aplikasi streaming dihentikan secara paksa atau node mati, alur recovery berlangsung seperti ini:

1. Driver baru dijalankan kembali (oleh cluster manager atau Kubernetes).
2. Spark membaca metadata dari checkpoint location.
3. Offset terakhir menjadi titik awal pembacaan ulang data dari sumber.
4. State terakhir dipulihkan, dan pemrosesan berlanjut dari sana.

restart driver → baca checkpoint → lanjut dari offset terakhir → proses ulang

#Exactly-Once vs At-Least-Once

Semantik pengiriman tergantung sink:

• Kafka: dengan sink Kafka dan checkpoint, Spark memberikan exactly-once — data tidak hilang dan tidak duplikat.
• File sink (Parquet): memberikan at-least-once; jika job gagal di tengah penulisan, batch bisa ditulis ulang.

spark-submit --conf spark.sql.streaming.schemaInference=true \
  --conf spark.sql.streaming.fileSink.log.deletion=true job.py

spark.sql.streaming.fileSink.log.deletion adalah salah satu konfigurasi pembersihan metadata file sink. Untuk jaminan duplikat yang lebih ketat di batch processing, kalian bisa memanfaatkan Delta Lake yang menyediakan transaksi idempoten.

#State Management untuk Streaming Jobs

#State dari Window dan Aggregasi

Setiap agregasi ber-state (window, groupBy) menyimpan hasil sementara di memory. Watermark membantu membersihkan state yang sudah tidak relevan. Jika state membesar tanpa kendali, aplikasi bisa melambat atau memori meluap.

#State API: mapGroupsWithState

Untuk logika state kustom — misalnya session tracking — Spark menyediakan API mapGroupsWithState dan flatMapGroupsWithState:

flatMapGroupsWithState di Scala

import org.apache.spark.sql.streaming.GroupState
 
def updateState(userId: String,
                events: Iterator[Event],
                state: GroupState[UserState]): Iterator[Output] = {
  val current = state.getOption.getOrElse(UserState.empty)
  val updated = current.update(events)
  state.update(updated)
  Iterator(Output(userId, updated.summary))
}

GroupState[UserState] menyimpan state per kunci yang dipersist secara periodik ke checkpoint. Wajib menentukan timeout agar state kunci yang tidak aktif bisa dihapus — kalau tidak, state hanya akan tumbuh.

#Membersihkan State dengan Timeout

Mengatur timeout state

state.setTimeoutDuration("30 minutes")

state.setTimeoutDuration("30 minutes") otomatis memicu callback timeout untuk kunci yang tidak menerima event selama 30 menit — pola standar untuk sessionization.

#Best Practices untuk Long-Running Streaming

#Desain yang Tahan Lama

Beberapa praktik yang menjaga aplikasi streaming tetap sehat selama berbulan-bulan:

• Checkpoint di storage persisten: bukan disk lokal yang bisa hilang bersama pod.
• Monitor lag dan state size: alert saat lag Kafka membesar atau state melewati ambang.
• Restart yang bersih: pastikan shutdown memakai query.stop() dan driver diberi waktu menyelesaikan penulisan batch terakhir.
• Jangan ubah struktur query sembarangan saat checkpoint sudah ada.

#Menangani Restart dan Upgrade

Saat meng-upgrade kode atau library:

1. submit query baru dengan checkpoint baru (mode uji)
2. verifikasi data tidak ganda dan sink normal
3. cutover: hentikan query lama, jalankan yang baru

Pola ini menghindari kegagalan di tengah malam karena perubahan yang belum teruji terhadap format checkpoint lama.

#Penutup

Episode 16 menuntaskan aspek keandalan: checkpoint menyimpan offset, state, dan metadata agar job bisa lanjut setelah kegagalan; recovery memulihkan posisi dan state secara otomatis; state management dengan mapGroupsWithState mengontrol logika kustom; dan best practices menjaga aplikasi streaming tetap sehat dalam jangka panjang.

Inti yang harus dibawa pulang:

• Checkpoint location harus di storage persisten dan konsisten.
• Offset log di checkpoint mencegah kehilangan dan duplikasi data.
• Semantik exactly-once tergantung sink — Kafka mendukung, file sink at-least-once.
• State kustom membutuhkan timeout eksplisit agar tidak membesar.
• Upgrade struktur query sebaiknya memakai checkpoint baru.

Di episode 17 selanjutnya kita akan membahas custom extensions dan UDFs — menulis UDF, UDAF, dan UDT di Scala, Java, dan Python, custom serialization dengan Kryo, mengekstensi Spark dengan library eksternal, serta packaging library untuk dipakai ulang di banyak job.