Delta Lake: Penyimpanan Lakehouse yang Andal dengan Transaksi ACID

Data lake yang dibangun di atas object storage memang murah dan mudah diskalakan, tetapi format file biasa tidak memberi jaminan apa pun: penulisan yang gagal di tengah jalan bisa membuat pembaca menerima data rusak, dan job yang berjalan bersamaan bisa saling menimpa. Delta Lake mengatasi ini dengan menambahkan transaction log di atas Parquet, sehingga data lake memperoleh transaksi ACID, penegakan skema, dan time travel. Tutorial ini membahas Delta Lake dari dasar melalui dua jalur praktis: library Python deltalake yang ringan (tanpa Spark) dan Apache Spark dengan delta-spark.

Apa Itu Delta Lake

Delta Lake adalah format tabel terbuka (open table format). Sebuah tabel Delta bukan satu file, melainkan sebuah direktori yang berisi:

• Satu atau lebih file data Parquet yang menyimpan baris data sebenarnya.
• Sebuah direktori transaction log bernama deltalog, berisi file commit JSON yang berurutan (00000000000000000000.json, 00000000000000000001.json, dan seterusnya) ditambah checkpoint Parquet berkala.

Setiap perubahan pada tabel dicatat sebagai commit atomik di dalam log. Pembaca merujuk pada log untuk mengetahui persis file Parquet mana yang membentuk versi tabel saat ini. Karena log adalah sumber kebenaran, pembaca tidak akan pernah melihat file yang ditulis sebagian, dan dua penulis dapat mendeteksi bahwa mereka berkonflik.

Inilah inti dari ide lakehouse: mempertahankan biaya rendah dan keterbukaan data lake (Parquet di S3, ADLS, GCS, atau disk lokal), tetapi menambahkan keandalan dan semantik transaksional seperti yang diharapkan dari sebuah data warehouse.

Mengapa Tidak Cukup dengan Parquet Saja?

Parquet biasa adalah penyimpanan kolumnar yang sangat baik, tetapi tidak memiliki konsep versi tabel atau commit. Jika sebuah job menulis sepuluh file Parquet lalu gagal setelah lima file, pembaca yang memindai direktori akan mengambil campuran data yang tidak konsisten. Tidak ada "append" yang atomik, tidak ada update bersamaan yang aman, tidak ada cara untuk rollback, dan perubahan skema sepenuhnya manual. Delta Lake tetap memakai Parquet sebagai lapisan penyimpanan dan menambahkan metadata transaksional di atasnya.

Transaction Log Secara Rinci

Direktori Delta yang umum terlihat seperti ini:

orders/
  deltalog/
    00000000000000000000.json
    00000000000000000001.json
    00000000000000000002.json
    lastcheckpoint
  part-00000-....snappy.parquet
  part-00001-....snappy.parquet

Setiap commit JSON menjelaskan action: file ditambahkan (add), file dihapus (remove), perubahan metadata, dan perubahan protokol. Versi 0 adalah pembuatan tabel; setiap commit berikutnya menambah nomor versi sebanyak satu. Nomor versi yang monoton inilah yang memungkinkan time travel, dan action remove itulah yang nantinya memungkinkan VACUUM mengembalikan ruang disk.

Jalur 1: Library Python deltalake (Tanpa Spark)

Paket deltalake (proyek delta-rs) adalah implementasi native Rust dengan binding Python. Paket ini membaca dan menulis tabel Delta tanpa JVM atau klaster Spark, sehingga ideal untuk ETL single-node, notebook, dan menyematkan Delta ke dalam pipeline pandas atau PyArrow yang sudah ada.

Instalasi

pip install "deltalake>=0.18" pandas pyarrow

Membuat dan Menambahkan Data ke Tabel

Kita akan memodelkan tabel orders sepanjang tutorial ini.

import pandas as pd
from deltalake import writedeltalake, DeltaTable


tablepath = "data/orders"

orders = pd.DataFrame(
    {
        "orderid": [1, 2, 3],

        "customer": ["alice", "bob", "carol"],
        "amount": [120.0, 80.5, 45.0],
        "status": ["paid", "paid", "pending"],
        "region": ["west", "east", "west"],
    }
)

Penulisan pertama membuat tabel pada versi 0.
writedeltalake(tablepath, orders, mode="overwrite")

neworders = pd.DataFrame(
    {
        "orderid": [4, 5],

        "customer": ["dave", "erin"],
        "amount": [200.0, 15.0],
        "status": ["paid", "paid"],
        "region": ["east", "west"],
    }
)

writedeltalake(tablepath, neworders, mode="append")

Membaca Tabel

dt = DeltaTable(tablepath)

Sebagai DataFrame pandas
df = dt.topandas()
print(df)

Sebagai tabel PyArrow (berguna untuk dialirkan ke engine lain)
arrowtable = dt.topyarrowtable()


Memeriksa versi saat ini
print("versi saat ini:", dt.version())

paid = dt.topandas(
    columns=["orderid", "customer", "amount"],

    filters=[("status", "=", "paid")],
)

# Ini akan memunculkan error karena discount adalah kolom baru.
bad = pd.DataFrame({"orderid": [6], "discount": [0.1]})
writedeltalake(tablepath, bad, mode="append")  # -> error skema tidak cocok

Ketika perubahan skema memang disengaja, aktifkan secara eksplisit dengan schemamode="merge":

extended = pd.DataFrame(
    {
        "orderid": [6],
        "customer": ["frank"],
        "amount": [99.0],
        "status": ["paid"],
        "region": ["east"],
        "discount": [0.10],  # kolom yang benar-benar baru
    }
)

writedeltalake(tablepath, extended, mode="append", schemamode="merge")

Baris yang sudah ada akan mendapat null untuk kolom discount yang baru. Gunakan schemamode="overwrite" bersama mode="overwrite" hanya ketika Anda memang ingin mengganti skema.

Time Travel dengan deltalake

Karena setiap commit memiliki versi, Anda dapat memuat keadaan historis tabel kapan pun.

# Memuat versi tertentu
v0 = DeltaTable(tablepath, version=0).topandas()

Atau memuat berdasarkan versi
dt = DeltaTable(tablepath)

dt.loadasversion(1)

snapshot = dt.topandas()

Memeriksa seluruh riwayat
for commit in DeltaTable(tablepath).history():

    print(commit.get("version"), commit.get("operation"), commit.get("timestamp"))

dt = DeltaTable(tablepath)

DELETE baris yang cocok dengan predikat
dt.delete("status = 'pending'")

UPDATE baris
dt.update(
    predicate="region = 'west'",
    updates={"status": "'shipped'"},
)

Upsert menggunakan API merge. Polanya adalah: mencocokkan berdasarkan kunci, melakukan update saat cocok, dan insert saat tidak cocok.

import pyarrow as pa

updates = pa.table(
    {
        "orderid": [4, 7],          # 4 sudah ada -> update; 7 baru -> insert

        "customer": ["dave", "gina"],
        "amount": [222.0, 60.0],
        "status": ["refunded", "paid"],
        "region": ["east", "west"],
    }
)

(
    DeltaTable(tablepath)
    .merge(
        source=updates,
        predicate="target.orderid = source.orderid",
        sourcealias="source",

        targetalias="target",
    )
    .whenmatchedupdateall()

    .whennotmatchedinsertall()

    .execute()
)

dt = DeltaTable(tablepath)

Menggabungkan file kecil menjadi file yang lebih besar
dt.optimize.compact()

Menghapus file data yang tidak lagi dirujuk oleh log.
retentionhours menjaga agar file yang masih dibutuhkan time travel tidak terhapus.

dt.vacuum(retentionhours=168, dryrun=False, enforceretentionduration=True)

pip install pyspark==3.5.1 delta-spark==3.2.0

from pyspark.sql import SparkSession
from delta import configuresparkwithdeltapip


builder = (
    SparkSession.builder.appName("delta-orders")
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
    .config(
        "spark.sql.catalog.sparkcatalog",
        "org.apache.spark.sql.delta.catalog.DeltaCatalog",
    )
)

spark = configuresparkwithdeltapip(builder).getOrCreate()

Membuat dan Menambahkan Data dengan Spark

data = [
    (1, "alice", 120.0, "paid", "west"),
    (2, "bob", 80.5, "paid", "east"),
    (3, "carol", 45.0, "pending", "west"),
]
columns = ["orderid", "customer", "amount", "status", "region"]


df = spark.createDataFrame(data, columns)

Menulis tabel Delta yang dipartisi
(
    df.write.format("delta")
    .mode("overwrite")
    .partitionBy("region")
    .save("/data/ordersspark")
)

Menambah baris
more = spark.createDataFrame([(4, "dave", 200.0, "paid", "east")], columns)
more.write.format("delta").mode("append").save("/data/ordersspark")

CREATE TABLE orders
USING DELTA
LOCATION '/data/ordersspark';

Evolusi Skema dengan mergeSchema

from pyspark.sql.functions import lit

evolved = (
    spark.createDataFrame([(5, "erin", 30.0, "paid", "west")], columns)
    .withColumn("discount", lit(0.05))
)

(
    evolved.write.format("delta")
    .mode("append")
    .option("mergeSchema", "true")
    .save("/data/ordersspark")

)

# Berdasarkan versi
v0 = spark.read.format("delta").option("versionAsOf", 0).load("/data/ordersspark")

Berdasarkan timestamp
ts = (
    spark.read.format("delta")
    .option("timestampAsOf", "2026-06-01 00:00:00")
    .load("/data/ordersspark")

)

SELECT  FROM orders VERSION AS OF 0;
SELECT  FROM orders TIMESTAMP AS OF '2026-06-01 00:00:00';
DESCRIBE HISTORY orders;

MERGE INTO orders AS target
USING updates AS source
ON target.orderid = source.orderid

WHEN MATCHED AND source.status = 'cancelled' THEN
  DELETE
WHEN MATCHED THEN
  UPDATE SET
    target.amount = source.amount,
    target.status = source.status
WHEN NOT MATCHED THEN
  INSERT (orderid, customer, amount, status, region)
  VALUES (source.orderid, source.customer, source.amount, source.status, source.region);

from delta.tables import DeltaTable as SparkDeltaTable

target = SparkDeltaTable.forPath(spark, "/data/ordersspark")
source = spark.createDataFrame(
    [(4, "dave", 222.0, "refunded", "east")], columns
)

(
    target.alias("t")
    .merge(source.alias("s"), "t.orderid = s.orderid")
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute()
)

DELETE dan UPDATE di Spark

DELETE FROM orders WHERE status = 'pending';

UPDATE orders SET status = 'shipped' WHERE region = 'west';

OPTIMIZE, Z-Ordering, dan VACUUM

-- Menggabungkan file kecil
OPTIMIZE orders;

-- Menempatkan data terkait berdekatan untuk mempercepat pembacaan terfilter
OPTIMIZE orders ZORDER BY (customer);

-- Mengembalikan ruang dari file yang lebih tua dari rentang retensi
VACUUM orders RETAIN 168 HOURS;

Z-ordering menata ulang data di dalam file sehingga baris dengan nilai yang mirip pada kolom Z-order berada berdekatan. Ketika Anda nantinya memfilter berdasarkan kolom tersebut, Delta dapat melewati lebih banyak file melalui statistik data-skipping. Ini paling berguna untuk kolom berkardinalitas tinggi yang sering Anda filter, seperti customer atau orderid.

from deltalake import DeltaTable

df = DeltaTable("/data/ordersspark").topandas(filters=[("region", "=", "west")])

INSTALL delta;
LOAD delta;

SELECT region, count(*) AS n, sum(amount) AS total
FROM deltascan('/data/ordersspark')

GROUP BY region
ORDER BY total DESC;

• Partisi berdasarkan kolom berkardinalitas rendah yang sering Anda filter (misalnya region atau tanggal event), dan hindari partisi berlebihan yang menciptakan banyak file mungil.
• Gunakan MERGE INTO untuk ingest yang idempoten agar pemrosesan ulang batch yang sama tidak menghasilkan duplikat.
• Jalankan OPTIMIZE secara berkala pada tabel yang menerima banyak append kecil, dan tambahkan ZORDER BY pada kolom yang paling sering Anda filter.
• Tetapkan retensi VACUUM yang masuk akal (default-nya 7 hari). Jangan menurunkannya di bawah jendela time-travel terpanjang yang Anda andalkan, atau pembacaan historis akan rusak.
• Aktifkan evolusi skema secara eksplisit dengan mergeSchema / schemamode="merge" alih-alih mengaktifkannya secara global, sehingga pergeseran skema yang tidak disengaja tetap terdeteksi.
• Jaga keselarasan versi penulis: dengan delta-spark, biarkan configuresparkwithdeltapip mengelola versi paket; dengan deltalake, kunci versi minimum di requirements Anda.
• Pilih delta-rs untuk job single-node dan Spark untuk beban terdistribusi; keduanya membaca dan menulis format on-disk yang sama, jadi Anda dapat menggabungkannya.

Kesimpulan dan Poin Penting

Delta Lake mengubah direktori berisi file Parquet menjadi tabel yang andal dan transaksional dengan menambahkan deltalog yang berversi. Poin-poin penting yang perlu diingat:

• Tabel Delta adalah data Parquet ditambah transaction log; log adalah sumber kebenaran dan dasar jaminan ACID.
• Anda memiliki dua titik masuk praktis: library Python deltalake untuk pekerjaan single-node tanpa Spark, dan delta-spark untuk pemrosesan terdistribusi dan SQL.
• Penegakan skema melindungi Anda secara default; aktifkan evolusi secara sengaja.
• Time travel memungkinkan Anda membaca versi lampau mana pun berdasarkan nomor atau timestamp, sangat berharga untuk debugging dan reproduksibilitas.
• MERGE, UPDATE, dan DELETE adalah operasi kelas utama, dan OPTIMIZE/Z-ordering ditambah VACUUM menjaga tabel tetap cepat dan rapi.
• Dibandingkan Parquet biasa, Delta menambah keandalan dengan overhead kecil; dibandingkan Iceberg dan Hudi, ia merupakan pilihan kuat untuk tim yang berpusat pada Spark dan Python.

Dengan blok bangunan ini, Anda dapat menjalankan lakehouse yang andal di atas object storage komoditas, dan membaca tabel yang sama dari Spark, pandas, atau DuckDB seiring berkembangnya kebutuhan Anda.