Delta Lake

ACID di atas data lake, time travel, posisi MERGE.

Daftar isi

Format berkas pada modul berkas dan format menulis data dengan murah dan cepat, tetapi tidak memberi jaminan transaksi. Bila dua pekerjaan menulis ke folder Parquet yang sama pada waktu bersamaan, hasilnya bisa campur aduk, dan pembaca yang datang di tengah penulisan bisa membaca setengah data. Delta Lake menutup celah ini dengan menambahkan lapisan transaksi di atas berkas Parquet, dan topik ini memeragakan jaminan tersebut beserta time travel dan MERGE pada tabel Delta yang sungguhan.

Lapisan transaksi di atas Parquet#

Tabel Delta adalah folder Parquet ditambah satu subfolder bernama _delta_log. Data baris tetap tersimpan sebagai berkas Parquet biasa yang bisa dibaca alat mana pun, sedangkan _delta_log menyimpan urutan commit yang menjelaskan berkas mana yang menjadi bagian tabel pada tiap versi. Setiap penulisan menjadi satu commit atomik, yaitu menulis berkas data baru lalu menambahkan satu entri log yang sah, sehingga pembaca pada titik mana pun selalu melihat keadaan tabel yang konsisten.

Jaminan ini disebut ACID dalam artian yang wajar untuk data lake. Atomisitas datang dari entri log tunggal, konsistensi datang dari pemeriksaan schema dan batasan pada commit, isolasi datang dari transaksi optimistik yang memeriksa konflik sebelum commit, dan keawetan datang dari Parquet yang sudah tersimpan. Penulis yang saling bertabrakan tidak menghancurkan tabel, melainkan salah satunya gagal dan boleh diulang, sebagaimana basis data transaksional pada umumnya.

Di sisi Spark, Delta menyatu sebagai format dan sebagai katalog. Format delta menggantikan parquet pada read dan write, sementara sesi perlu dua konfigurasi, yaitu extension io.delta.sql.DeltaSparkSessionExtension dan katalog org.apache.spark.sql.delta.catalog.DeltaCatalog, agar perintah SQL seperti UPDATE dan MERGE dikenali. Pada PySpark, paket pip delta-spark menyediakan helper configure_spark_with_delta_pip yang merakit jar Delta dari Maven saat sesi dibangun.

Satu tabel Delta yang diperbarui dan digabungkan#

Skrip berikut membuat tabel Delta berisi tiga baris nilai ujian, memperbarui satu baris melalui DeltaTable.update, menggabungkan perubahan melalui MERGE, membaca riwayat versi, dan menutup dengan time travel ke versi nol. Eksekusinya di sandbox melewati batas verifikasi otomatis karena inisialisasi jar Delta dan log transaksinya, sehingga keluarannya diambil dari pengujian manual yang skripnya identik.

python delta_dasar.py

# tidak-dijalankan
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("delta-dasar")
    .config("spark.sql.shuffle.partitions", "2")
    .config("spark.sql.debug.maxToStringFields", "500")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
    .getOrCreate()
)

lokasi = "/tmp/delta-nilai"
import shutil
shutil.rmtree(lokasi, ignore_errors=True)

df = spark.createDataFrame(
    [(1, "Ali", 85), (2, "Budi", 90), (3, "Citra", 78)],
    "id long, nama string, nilai long",
)
df.write.format("delta").mode("overwrite").save(lokasi)

from delta import DeltaTable

dt = DeltaTable.forPath(spark, lokasi)

print("=== isi tabel ===")
spark.read.format("delta").load(lokasi).orderBy("id").show()

print("=== UPDATE dan baca ulang ===")
dt.update(F.col("nama") == "Citra", {"nilai": F.lit(88)})
spark.read.format("delta").load(lokasi).orderBy("id").show()

print("=== MERGE: perbarui yang ada, sisipkan yang baru ===")
perubahan = spark.createDataFrame(
    [(2, "Budi", 95), (4, "Dewi", 92)],
    "id long, nama string, nilai long",
)
(
    dt.alias("tabel")
    .merge(perubahan.alias("sumber"), "tabel.id = sumber.id")
    .whenMatchedUpdate(set={"nilai": "sumber.nilai"})
    .whenNotMatchedInsert(values={"id": "sumber.id", "nama": "sumber.nama", "nilai": "sumber.nilai"})
    .execute()
)
spark.read.format("delta").load(lokasi).orderBy("id").show()

print("=== riwayat versi ===")
for baris in dt.history().select("version", "operation").collect():
    print("versi", baris["version"], "->", baris["operation"])

print("=== time travel ke versi 0 ===")
spark.read.format("delta").option("versionAsOf", 0).load(lokasi).orderBy("id").show()

spark.stop()

text Ilustrasi keluaran

WARNING: Using incubator modules: jdk.incubator.vector
=== isi tabel ===
+---+-----+-----+
| id| nama|nilai|
+---+-----+-----+
|  1|  Ali|   85|
|  2| Budi|   90|
|  3|Citra|   78|
+---+-----+-----+

=== UPDATE dan baca ulang ===
+---+-----+-----+
| id| nama|nilai|
+---+-----+-----+
|  1|  Ali|   85|
|  2| Budi|   90|
|  3|Citra|   88|
+---+-----+-----+

=== MERGE: perbarui yang ada, sisipkan yang baru ===
+---+-----+-----+
| id| nama|nilai|
+---+-----+-----+
|  1|  Ali|   85|
|  2| Budi|   95|
|  3|Citra|   88|
|  4| Dewi|   92|
+---+-----+-----+

=== riwayat versi ===
versi 2 -> MERGE
versi 1 -> UPDATE
versi 0 -> WRITE

=== time travel ke versi 0 ===
+---+-----+-----+
| id| nama|nilai|
+---+-----+-----+
|  1|  Ali|   85|
|  2| Budi|   90|
|  3|Citra|   78|
+---+-----+-----+

Tabel keluaran menceritakan seluruh siklus transaksi. update mengubah Citra menjadi 88 dan tercatat sebagai versi 1, lalu merge memperbarui Budi menjadi 95 sekaligus menyisipkan Dewi dalam satu commit versi 2, tanpa dua perintah terpisah. Riwayat versi merekam operasi per versi, dan bacaan dengan versionAsOf 0 mengembalikan isi asli, termasuk nilai Citra yang masih 78 sebelum diubah. Semua ini terjadi tanpa salinan manual dan tanpa skrip pemulihan, karena setiap versi tersimpan oleh log itu sendiri.

Perhatikan pula perilaku MERGE yang menjadikannya inti pipeline berlapis, yaitu sisi sumber dipasangkan dengan kunci, baris yang cocok diperbarui, dan baris yang belum ada disisipkan. Pola ini menggantikan alur baca penuh, bandingkan, tulis ulang yang rawan konflik pada folder Parquet polos, dan karenanya menjadi bawaan pada hampir semua pipeline upsert.

Posisi Delta dalam arsitektur data#

Delta menempati lapisan penyimpanan analitik pada pola lakehouse, yaitu menyatukan fleksibilitas data lake dengan jaminan gudang data. Pada pola medallion yang dibahas pada modul pengerjaan proyek, tabel bronze, silver, dan gold biasanya berupa tabel Delta karena setiap tahap butuh penulisan transaksional, pembaruan data yang tiba terlambat, dan pembacaan konsisten oleh tahap berikutnya. Struktur streaming pada modul streaming juga menyambung ke Delta sebagai sink dengan jaminan yang sama.

Perbandingan dengan format tetangganya membantu menempatkan pilihan. Apache Iceberg dan Apache Hudi menyediakan jaminan tabel serupa dengan pendekatan log yang berbeda, dan ketiganya kini dapat dibaca Spark. Delta unggul pada kedalaman integrasinya dengan Spark, termasuk perintah SQL-nya dan checkpoint streaming yang menyatu, sehingga bagi pembelajar Spark, Delta menjadi pilihan paling mulus sebagai tabel pertama.

Kesalahan umum#

Kesalahan pertama adalah menulis tabel Delta tanpa konfigurasi extension dan katalog, lalu bingung karena save berhasil tetapi perintah SQL Delta ditolak. Format berkas tetap bisa ditulis tanpa extension, sedangkan DML seperti UPDATE dan MERGE menuntut keduanya. Sertakan kedua konfigurasi pada builder, atau gunakan helper pip yang menyiapkannya.

Kesalahan kedua adalah menganggap folder _delta_log sebagai sampah yang boleh dihapus atau folder tabel yang boleh dipindah sebagiannya. Log adalah kebenaran tabel, dan menghapus entri log membuat versi hilang atau tabel tak terbaca. Pembersihan versi lama dilakukan dengan VACUUM dan retensi yang dinyatakan, bukan dengan menghapus berkas manual.

Kesalahan ketiga adalah memperlakukan time travel sebagai cadangan permanen. Setiap penulisan menambah versi, log membesar, dan versi yang terlalu lama akhirnya di-vacuum. Untuk kebutuhan arsip jangka panjang, salinan tabel atau retensi yang disengaja tetap diperlukan, dan time travel dirancang untuk pembatalan segera serta pengujian ulang, bukan pengganti cadangan.

Ringkasan#

Tabel Delta adalah folder Parquet ditambah _delta_log yang mencatat commit atomik, sehingga penulisan bersamaan tidak menghasilkan data campur aduk.
DeltaTable menyediakan update dan merge yang menulis commit transaksional, serta history yang membaca riwayatnya, dan time travel membaca versi lampau dengan versionAsOf.
Konfigurasi extension dan katalog menyatukan DML Delta ke SQL Spark, dan helper pip menyediakan jar Delta saat sesi dibangun.
Delta menempati lapisan tabel pada pola lakehouse dan medallion, dengan Iceberg dan Hudi sebagai tetangga yang jaminannya serupa.

Referensi#

Delta Lake on Apache Spark — Delta Lake documentation — Pengenalan resmi arsitektur log transaksi dan fitur tabel Delta.
Table batch reads and writes — Delta Lake documentation — Referensi write, read, update, merge, dan time travel.
delta-spark on PyPI — Paket pip resmi Delta Lake beserta helper configure_spark_with_delta_pip.