Chapter Spark Structured Streaming pada section Pemrosesan Data Skala Besar telah memperkenalkan model micro-batch, window, watermark, dan output mode. Chapter ini turun satu lapisan ke pertanyaan yang menentukan pipeline produksi: apa yang terjadi di antara dua batch? Hitungan per window disimpan di mana, bagaimana aplikasi yang direstart melanjutkan tepat dari titik berhenti, dan bagaimana hasil ditulis ke hilir tanpa duplikat. Ketiganya kita praktikkan dalam satu eksekusi nyata yang memuat state store, checkpoint, stream-static join, dan foreachBatch sekaligus.

State Store: Ingatan Antar Batch #

Agregasi berbasis window bersifat stateful: jumlah pesanan per window per produk harus terus diingat antar batch karena baris window yang sama datang tersebar di banyak batch. Structured Streaming menyimpan ingatan itu di state store — penyimpanan berkunci yang berversi per batch, diisi oleh hasil agregasi setiap micro-batch, dan dibersihkan ketika watermark menutup window sebagaimana dibuktikan pada Chapter Watermarks dan Late Events. Tanpa watermark, state tidak pernah menyusut dan memori tumbuh sepanjang usia pipeline.

Checkpoint adalah pasangan wajib state store: direktori yang merekam offset sumber tiap batch, keputusan commit, dan snapshot state. Aplikasi yang crash lalu direstart dengan checkpointLocation yang sama akan melanjutkan dari offset terakhir yang tercatat, memuat state terakhir, dan tidak mengulang batch yang sudah berkomit.

Demo: Satu Skrip, Empat Mekanisme #

Skrip berikut menghitung pesanan per window 8 detik per produk, memperkaya hasilnya dengan kategori dari tabel dimensi statis, mencatat checkpoint, lalu menutup dengan kueri kedua yang memakai foreachBatch:

python 
"""Stateful streaming + checkpoint: agregasi window diperkaya dimensi statis."""
import os
import shutil
import tempfile
import time

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder
         .master("local[1]")
         .config("spark.ui.enabled", "false")
         .config("spark.sql.shuffle.partitions", "2")
         .getOrCreate())
spark.sparkContext.setLogLevel("ERROR")

t0 = time.time()

# Dimensi statis kecil: kategori per produk (tabel dimensi).
kategori_df = spark.createDataFrame(
    [("Rok Tenun", "pakaian"), ("Kain Songket", "pakaian"),
     ("Tas Rotan", "aksesoris"), ("Blouse Katun", "pakaian")],
    "produk STRING, kategori STRING")

daftar = F.array(F.lit("Rok Tenun"), F.lit("Kain Songket"),
                 F.lit("Tas Rotan"), F.lit("Blouse Katun"))

# Stream: 3 pesanan/detik, event_time = waktu kedatangan.
pesanan = (spark.readStream.format("rate")
           .option("rowsPerSecond", 3)
           .option("numPartitions", 1)
           .load()
           .withColumn("produk", F.element_at(daftar, (F.col("value") % 4 + 1).cast("int")))
           .withColumn("event_time", F.col("timestamp")))

# Kueri 1: agregasi stateful (window + watermark), lalu stream-static join
# untuk memperkaya hasil agregasi dengan kolom kategori dari dimensi statis.
ringkasan = (pesanan
             .withWatermark("event_time", "4 seconds")
             .groupBy(F.window("event_time", "8 seconds"), F.col("produk"))
             .count())

laporan = (ringkasan.join(kategori_df, "produk")
           .select(F.col("window.start").alias("mulai"),
                   F.col("window.end").alias("selesai"),
                   "produk", "kategori", "count"))

ckpt = tempfile.mkdtemp(prefix="ckpt_demo_")
print(f"lokasi checkpoint: {ckpt}")

q = (laporan.writeStream
     .format("console")
     .outputMode("append")
     .trigger(processingTime="5 seconds")
     .option("truncate", "false")
     .option("checkpointLocation", ckpt)
     .start())
q.awaitTermination(32)
q.stop()

print("== isi direktori checkpoint ==")
for nama in sorted(os.listdir(ckpt)):
    jalur = os.path.join(ckpt, nama)
    if os.path.isdir(jalur):
        isi = sorted(os.listdir(jalur))
        print(f"  {nama}/ ({len(isi)} entri: {', '.join(isi[:6])})")
    else:
        print(f"  {nama}")

# Kueri 2: foreachBatch -- sink kustom menerima micro-batch + batch_id.
ckpt2 = tempfile.mkdtemp(prefix="ckpt_feb_")


def catat(df, batch_id):
    print(f"foreachBatch batch {batch_id}: {df.count()} baris")


q2 = (pesanan.writeStream
      .foreachBatch(catat)
      .trigger(processingTime="4 seconds")
      .option("checkpointLocation", ckpt2)
      .start())
q2.awaitTermination(12)
q2.stop()

shutil.rmtree(ckpt, ignore_errors=True)
shutil.rmtree(ckpt2, ignore_errors=True)
print(f"durasi script: {time.time() - t0:.1f} detik")
spark.stop()

Output yang Dihasilkan (Batch 0 sampai 3 mencetak tabel kosong — mode append menunggu watermark melewati akhir window, persis fenomena Chapter Spark Structured Streaming — dan Batch 5 sampai 7 berpola sama dengan Batch 4; keduanya dipangkas agar ringkas):

text 
lokasi checkpoint: /tmp/ckpt_demo_lzpcfbg2

-------------------------------------------
Batch: 4
-------------------------------------------
+-------------------+-------------------+------------+---------+-----+
|mulai              |selesai            |produk      |kategori |count|
+-------------------+-------------------+------------+---------+-----+
|2026-10-16 14:10:08|2026-10-16 14:10:16|Blouse Katun|pakaian  |2    |
|2026-10-16 14:10:08|2026-10-16 14:10:16|Kain Songket|pakaian  |2    |
|2026-10-16 14:10:08|2026-10-16 14:10:16|Rok Tenun   |pakaian  |2    |
|2026-10-16 14:10:08|2026-10-16 14:10:16|Tas Rotan   |aksesoris|2    |
+-------------------+-------------------+------------+---------+-----+

== isi direktori checkpoint ==
  .metadata.crc
  commits/ (16 entri: .0.crc, .1.crc, .2.crc, .3.crc, .4.crc, .5.crc)
  metadata
  offsets/ (16 entri: .0.crc, .1.crc, .2.crc, .3.crc, .4.crc, .5.crc)
  sources/ (1 entri: 0)
  state/ (1 entri: 0)
foreachBatch batch 0: 0 baris
foreachBatch batch 1: 6 baris
foreachBatch batch 2: 12 baris
foreachBatch batch 3: 12 baris

durasi script: 46.9 detik

Mari kita telaah baris-baris kode di atas:

  1. Baris 20-23 (kategori_df = ...): tabel dimensi statis — pola stream-static join dari Chapter Stream Joins dipakai di sini pada hasil agregasi, bukan event mentah: setiap baris laporan yang keluar dari state langsung diperkaya kategori.
  2. Baris 38-41 (ringkasan = ...): agregasi stateful dengan watermark 4 detik per window 8 detik per produk — inilah operasi yang mengisi state store.
  3. Baris 43-46 (laporan = ringkasan.join(...)): join ke tabel statis tanpa watermark, karena sisi dimensi selalu lengkap.
  4. Baris 51-57 (writeStream ... checkpointLocation): direktori ckpt merekam offset, commit, dan state kueri pertama.
  5. Baris 70-84 (foreachBatch(catat)): kueri kedua tidak memakai sink bawaan — fungsi catat(df, batch_id) dipanggil sekali per micro-batch dengan DataFrame batch dan nomor identitasnya.

Direktori checkpoint memperlihatkan anatomi pemulihan: offsets/ menyimpan rentang data yang dibaca tiap batch, commits/ menandai batch yang selesai, dan state/ menyimpan versi state store per partition. Enam belas entri pada offsets/ dan commits/ berasal dari 8 batch yang masing-masing punya berkas data dan berkas checksum .crc — jejak yang bisa diaudit untuk menjawab “batch berapa yang sedang diproses saat aplikasi berhenti”.

foreachBatch untuk Sink Kustom #

Sink bawaan (console, file, kafka) tidak selalu cukup — kadang hasil harus ditulis ke dua tujuan, didedup, atau lewat pustaka yang tidak tahu streaming. foreachBatch menjembatani itu: fungsi kita menerima DataFrame biasa per micro-batch sehingga seluruh API batch (termasuk df.write dan operasi JDBC) bisa dipakai. Argumen batch_id adalah kunci praktis untuk idempotensi: tabel hilir bisa menyimpan batch_id yang sudah diproses dan melewati penulisan ulang, sehingga batch yang diulang pasca-restart tidak menghasilkan duplikat — angka 0, 6, 12, 12 baris pada output adalah jumlah baris per batch pada kueri kedua tersebut.

Toleransi Kegagalan dan Exactly-Once #

Checkpoint menjamin Spark tidak memproses satu batch dua kali secara internal, tetapi janji exactly-once ujung-ke-ujung menuntut tiga syarat sekaligus. Pertama, sumber harus bisa dibaca ulang — Kafka memenuhi ini karena pesan bertahan selama retention dan offset bisa ditelusuri ulang. Kedua, komputasi harus deterministik — hasil yang sama dari data yang sama, tanpa pengacakan yang memengaruhi output. Ketiga, sink harus idempotent atau transaksional — menulis ulang batch yang sama tidak menambah baris baru, misalnya lewat upsert berkunci atau penyimpanan batch_id seperti di atas. Ketiga syarat ini adalah sisi engine dari pembahasan delivery semantics pada Chapter Delivery Semantics.

Rincian resmi mekanisme pemulihan dan daftar sink terdokumentasi di Structured Streaming Programming Guide .

Pipeline yang tahan gagal tetap tidak berarti sehat setiap saat — lag bisa tumbuh dan batch bisa melambat tanpa satupun proses gagal. Cara mengukur dan memasang ambang peringatan itulah pokok bahasan Chapter Monitoring Streaming Pipeline.