KAL
Menu

Checkpoint dan sink

Jalur exactly-once, mode output.

Daftar isi

Kueri streaming yang benar bukan hanya yang hasilnya benar saat berjalan, melainkan yang tetap benar setelah gagal dan dihidupkan ulang. Checkpoint adalah mekanisme Spark untuk itu, yaitu titik pemulihan yang mencatat posisi data dan state kueri, sehingga kueri yang dihidupkan ulang meneruskan pekerjaan alih-alih mengulangnya. Topik ini menutup modul streaming dengan demonstrasi pemulihan nyata pada sink berkas, sekaligus merangkum jaminan pengiriman yang ditawarkan.

Apa yang tersimpan di checkpoint#

Folder checkpoint menyimpan beberapa kumpulan informasi utama. Pertama, log offset pada subfolder offsets, yaitu posisi terakhir yang sudah diproses dari source, misalnya daftar berkas yang sudah ditangani. Kedua, subfolder commits yang mencatat micro-batch yang selesai dengan sukses. Ketiga, state kueri pada subfolder state, yang menyimpan hasil agregasi berjalan dan garis watermark bila kueri memakainya. Subfolder metadata dan sources yang ikut tampak pada keluaran menampung informasi kueri dan sumbernya, sedangkan state hanya terbentuk bila kueri menyimpan state agregasi.

Ketika kueri dihidupkan ulang dengan checkpoint yang sama, Spark membaca log tersebut, menemukan posisi terakhir, lalu memproses hanya data yang belum dicatat. Tanpa checkpoint, posisi ini hilang begitu kueri berhenti, dan data bisa diproses dua kali atau terlewat tanpa cara untuk mengetahuinya. Inilah sebabnya Spark menuntut checkpointLocation pada setiap kueri streaming dan tidak memberi opsi untuk melewatinya.

Satu aturan penting menyertai hal ini, yaitu folder checkpoint melekat pada definisi kueri. Mengubah transformasi, schema, atau source di atas checkpoint lama memicu error ketidakcocokan state, karena hasil tersimpan tidak lagi cocok dengan rencana baru. Saat eksperimen mengubah definisi kueri, mulailah dengan folder checkpoint baru.

Demonstrasi pemulihan pada sink parquet#

Skrip berikut menulis ke sink Parquet dengan checkpoint di folder terpisah. Trigger availableNow dijalankan dua kali, dipisahkan oleh penulisan satu berkas sumber baru. Putaran kedua memeragakan kueri yang dihidupkan ulang di produksi, dan isi sink dibaca ulang setelah setiap putaran.

python uji_checkpoint.py 
import os
import shutil

from pyspark.sql import SparkSession

shutil.rmtree("/tmp/sumber-ckpt", ignore_errors=True)
shutil.rmtree("/tmp/ckpt-parquet", ignore_errors=True)
shutil.rmtree("/tmp/sink-parquet", ignore_errors=True)
os.makedirs("/tmp/sumber-ckpt")
with open("/tmp/sumber-ckpt/part1.txt", "w") as f:
    f.write("kamar1,28.5\n")
    f.write("kamar2,30.1\n")

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("uji-checkpoint")
    .config("spark.sql.shuffle.partitions", "2")
    .config("spark.sql.adaptive.enabled", "false")
    .getOrCreate()
)

df = (
    spark.readStream
    .format("csv")
    .schema("kamar string, suhu double")
    .load("/tmp/sumber-ckpt")
)


def jalankan():
    q = (
        df.writeStream
        .format("parquet")
        .option("path", "/tmp/sink-parquet")
        .option("checkpointLocation", "/tmp/ckpt-parquet")
        .trigger(availableNow=True)
        .start()
    )
    q.awaitTermination()


jalankan()
print("=== isi sink setelah putaran pertama ===")
spark.read.parquet("/tmp/sink-parquet").orderBy("kamar", "suhu").show()
print("isi folder checkpoint:", [d for d in sorted(os.listdir("/tmp/ckpt-parquet")) if not d.startswith(".")])

with open("/tmp/sumber-ckpt/part2.txt", "w") as f:
    f.write("kamar1,27.9\n")
jalankan()
print("=== isi sink setelah query dijalankan ulang ===")
spark.read.parquet("/tmp/sink-parquet").orderBy("kamar", "suhu").show()

spark.stop()
text Keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
=== isi sink setelah putaran pertama ===
+------+----+
| kamar|suhu|
+------+----+
|kamar1|28.5|
|kamar2|30.1|
+------+----+

isi folder checkpoint: ['commits', 'metadata', 'offsets', 'sources']
=== isi sink setelah query dijalankan ulang ===
+------+----+
| kamar|suhu|
+------+----+
|kamar1|27.9|
|kamar1|28.5|
|kamar2|30.1|
+------+----+

Dua tabel di atas adalah bukti pemulihan yang bekerja. Setelah putaran kedua, sink memuat tiga baris, yaitu dua baris dari putaran pertama plus satu baris baru, tanpa penggandaan. Apabila putaran kedua memproses ulang seluruh folder sumber, kamar1 bernilai 28.5 akan muncul dua kali. Daftar folder checkpoint juga tampil, dan keempat subfolder itu yang menyimpan posisi berkas serta metadata yang membuat keluaran kedua dimungkinkan.

Nilai 27.9 pada baris pertama putaran kedua berasal dari berkas part2.txt yang ditulis di antara dua putaran. Source berkas mengurutkan berkas menurut nama saat memilih berkas baru, dan karena part2.txt menyusul part1.txt, urutan prosesnya jelas. Pemberian nama berkas yang berurutan karenanya bukan sekadar penamaan kosmetik, melainkan bagian dari disiplin source berkas.

Jaminan pengiriman dan pilihan sink#

Kombinasi checkpoint dengan sink berkas menghasilkan jaminan exactly-once dari sudut pandang pembaca sink, yaitu setiap baris sumber tercatat di keluaran tepat satu kali meski kueri gagal di tengah jalan. Log offset mencegah pemrosesan ganda, sedangkan keluaran micro-batch yang gagal sebelum commit tidak pernah tampil. Jaminan ini melekat pada pasangan source yang bisa diputar ulang dan sink yang mencatat, dan kedua syarat itu dipenuhi oleh berkas di kedua sisi.

Sink konsol dan memori yang dipakai untuk belajar tidak memberi jaminan itu, karena keduanya hanya menampilkan isi dan tidak ikut mencatat. Pada produksi, pilihan sink yang lazim adalah berkas Parquet atau Delta untuk arsip analitik, dan Apache Kafka untuk aliran pesan antar sistem. Sink Delta menambah kemampuan penggabungan dan pembacaan konsisten, yang dijelaskan pada modul ekosistem.

Kata exactly-once layak dibaca dengan hati-hati, karena jaminannya berlaku pada hasil akhir yang terlihat pembaca, bukan pada setiap langkah internal. Kegagalan di tengah micro-batch boleh saja memproses data dua kali secara internal, tetapi hanya keluaran yang sudah di-commit yang pernah tampak, dan itulah yang dijanjikan. Pemahaman batas jaminan ini mencegah kesimpulan keliru ketika merancang pipeline yang menulis ke sistem eksternal.

Kesalahan umum#

Kesalahan pertama adalah membersihkan folder sink tanpa membersihkan checkpoint, atau sebaliknya. Checkpoint mencatat berkas yang sudah diproses, sedangkan sink menyimpan hasilnya, dan ketidaksesuaian keduanya membuat data hilang atau berantakan. Saat eksperimen diulang dari nol, ketiga folder, yaitu sumber, sink, dan checkpoint, dihapus bersamaan seperti pada awal skrip.

Kesalahan kedua adalah memakai satu folder checkpoint untuk beberapa kueri yang berbeda. State kueri-kueri itu saling menimpa dan error ketidakcocokan muncul saat kueri dihidupkan ulang. Setiap kueri berhak atas satu folder checkpoint-nya sendiri, dan penamaannya sebaiknya mencerminkan nama pipeline.

Kesalahan ketiga adalah menulis ke folder sink yang sama dengan folder sumber. Berkas keluaran masuk ke pemantauan source dan diproses kembali sebagai data baru, sehingga pipeline memproses keluarannya sendiri kembali tanpa henti. Pisahkan direktori sejak desain, dan bila arsitektur memang menuntut rantai streaming ke streaming, gunakan sink yang memisahkan path secara eksplisit.

Ringkasan#

  • Folder checkpoint menyimpan posisi data, catatan commit, dan state kueri, sehingga kueri yang dihidupkan ulang meneruskan pekerjaan tanpa penggandaan.
  • Demonstrasi dua putaran membuktikan berkas lama tidak diproses ulang dan hanya data baru yang masuk sink.
  • Kombinasi source yang bisa diputar ulang dengan sink berkas menghasilkan jaminan exactly-once pada keluaran yang terlihat.
  • Satu folder checkpoint per kueri, dan sumber, sink, serta checkpoint yang akan direset harus dibersihkan bersamaan.

Referensi#