Delta Lake

Iceberg menyimpan pohon metadata dalam berkas Avro berlapis. Delta Lake memilih bentuk yang lebih sederhana dan sangat mudah diinspeksi: satu direktori _delta_log berisi berkas JSON bernomor urut, satu berkas untuk setiap transaksi. Delta Lake diciptakan di Databricks dan kini merupakan proyek terbuka dengan versi stabil 3.x; implementasi Rust-nya, delta-rs, memungkinkan kita menjalankan seluruh fitur inti — menulis, membaca, time travel, hingga evolusi schema — tanpa cluster Spark. Chapter ini adalah praktik paling panjang di section: setiap blok kode benar-benar dijalankan dan output-nya disalin apa adanya.

Tabel Delta Pertama dengan delta-rs #

Pustaka deltalake menyediakan dua pintu utama: write_deltalake() untuk menulis dan DeltaTable untuk membaca. Blok berikut membuat tabel pesanan, membacanya, lalu membedah isi berkas log transaksinya baris demi baris:

python
import json
import os
import shutil

import pyarrow as pa
from deltalake import DeltaTable, write_deltalake

BASE = "delta-demo/pesanan"
if os.path.exists(BASE):
    shutil.rmtree(BASE)

skema = pa.schema(
    [
        pa.field("id", pa.int64()),
        pa.field("nama", pa.string()),
        pa.field("jumlah", pa.int64()),
    ]
)


def tabel(rows):
    return pa.Table.from_arrays(
        [
            pa.array([r[0] for r in rows], pa.int64()),
            pa.array([r[1] for r in rows]),
            pa.array([r[2] for r in rows], pa.int64()),
        ],
        schema=skema,
    )


write_deltalake(BASE, tabel([(1, "Budi", 2), (2, "Siti", 1), (3, "Dewi", 5)]))

dt = DeltaTable(BASE)
isi = dt.to_pyarrow_table()
print("versi tabel:", dt.version())
print("jumlah baris:", isi.num_rows)
for baris in sorted(isi.to_pylist(), key=lambda r: r["id"]):
    print(f"  {baris['id']} {baris['nama']} jumlah={baris['jumlah']}")

print("berkas _delta_log:", os.listdir(os.path.join(BASE, "_delta_log")))
nama_log = os.path.join(BASE, "_delta_log", f"{dt.version():020d}.json")
for nomor, baris_json in enumerate(open(nama_log), start=1):
    aksi = json.loads(baris_json)
    jenis = list(aksi.keys())[0]
    isi_aksi = aksi[jenis]
    if jenis == "commitInfo":
        print(f"  baris {nomor}: commitInfo | operation={isi_aksi['operation']} | engine={isi_aksi['engineInfo']}")
    elif jenis == "protocol":
        print(f"  baris {nomor}: protocol | minReader={isi_aksi['minReaderVersion']} minWriter={isi_aksi['minWriterVersion']}")
    elif jenis == "metaData":
        kolom = [f["name"] for f in json.loads(isi_aksi["schemaString"])["fields"]]
        print(f"  baris {nomor}: metaData | kolom={kolom} | partition={isi_aksi['partitionColumns']}")
    elif jenis == "add":
        statistik = json.loads(isi_aksi["stats"])
        print(f"  baris {nomor}: add | file parquet baru | numRecords={statistik['numRecords']}")

Output yang Dihasilkan:

text
versi tabel: 0
jumlah baris: 3
  1 Budi jumlah=2
  2 Siti jumlah=1
  3 Dewi jumlah=5
berkas _delta_log: ['00000000000000000000.json']
  baris 1: commitInfo | operation=WRITE | engine=delta-rs:py-1.6.6
  baris 2: protocol | minReader=1 minWriter=2
  baris 3: metaData | kolom=['id', 'nama', 'jumlah'] | partition=[]
  baris 4: add | file parquet baru | numRecords=3

Mari kita telaah baris-baris kode di atas:

Baris 8-18 (BASE, skema, pembersihan): tabel Delta hanyalah sebuah direktori; blok pembuka menghapus direktori lama agar contoh selalu mulai dari bersih, lalu skema menetapkan tiga kolom tabel.
Baris 21-29 (tabel(rows)): satu helper pembangun tabel PyArrow sehingga tiga contoh di chapter ini menulis data dengan bentuk yang sama.
Baris 32 (write_deltalake): satu panggilan ini menghasilkan dua hal sekaligus — berkas Parquet di dalam direktori tabel dan commit 00000000000000000000.json di _delta_log.
Baris 34-39 (isi, sorted(...)): pembacaan dilakukan lewat to_pyarrow_table(); hasilnya diurutkan berdasarkan id karena urutan antar-berkas tidak dijamin.
Baris 43-56 (for nomor, baris_json ...): setiap baris di berkas log adalah satu objek JSON tunggal; blok ini mengurai jenis aksinya dan mencetak inti setiap aksi.

Anatomi _delta_log #

Nama berkas log adalah nomor versi yang di-pad nol sampai dua puluh digit, sehingga urutan nama berkas sama dengan urutan waktu commit. Setiap baris di dalamnya adalah satu aksi:

Aksi	Peran
`commitInfo`	cap waktu, nama operasi, mesin penulis (misal `delta-rs:py-1.6.6`), metrik
`protocol`	versi protokol minimum yang harus dipahami pembaca dan penulis
`metaData`	schema tabel, kolom partisi, konfigurasi
`add`	berkas Parquet baru beserta statistik: jumlah baris, nilai minimum/maksimum
`remove`	berkas yang tidak lagi dipakai versi ini (misal setelah overwrite)

Query membaca seluruh aksi add pada versi terbaru, menggabungkan daftarnya, lalu membuka berkas Parquet yang dirujuk. Ketika log memanjang, Delta menulis checkpoint — ringkasan Parquet dari seluruh log — sehingga pembaca baru tidak perlu membaca seluruh JSON dari versi nol; Spark Delta membuat checkpoint secara berkala, dan delta-rs menyediakan create_checkpoint() untuk hal yang sama.

Time Travel Antar Versi #

Menambahkan data membuat commit baru — versi tabel naik dari 0 ke 1 — sementara versi lama tetap bisa dibuka kapan saja. Di delta-rs 1.x, time travel dilakukan lewat parameter version saat membuat objek DeltaTable, atau load_as_version() yang juga menerima penanda waktu:

python
import os

import pyarrow as pa
from deltalake import DeltaTable, write_deltalake

BASE = "delta-demo/pesanan"

skema = pa.schema(
    [
        pa.field("id", pa.int64()),
        pa.field("nama", pa.string()),
        pa.field("jumlah", pa.int64()),
    ]
)


def tabel(rows):
    return pa.Table.from_arrays(
        [
            pa.array([r[0] for r in rows], pa.int64()),
            pa.array([r[1] for r in rows]),
            pa.array([r[2] for r in rows], pa.int64()),
        ],
        schema=skema,
    )


write_deltalake(BASE, tabel([(4, "Ani", 3), (5, "Rina", 2)]), mode="append")

dt = DeltaTable(BASE)
isi = dt.to_pyarrow_table()
print("versi baru:", dt.version(), "|", isi.num_rows, "baris")
for baris in sorted(isi.to_pylist(), key=lambda r: r["id"]):
    print(f"  {baris['id']} {baris['nama']} jumlah={baris['jumlah']}")

lama = DeltaTable(BASE, version=0)
isi_lama = lama.to_pyarrow_table()
print("time travel versi 0:", lama.version(), "|", isi_lama.num_rows, "baris")
print("  nama di versi 0:", [b["nama"] for b in sorted(isi_lama.to_pylist(), key=lambda r: r["id"])])

berkas_log = sorted(os.listdir(os.path.join(BASE, "_delta_log")))
print("berkas _delta_log:", berkas_log)

Output yang Dihasilkan:

text
versi baru: 1 | 5 baris
  1 Budi jumlah=2
  2 Siti jumlah=1
  3 Dewi jumlah=5
  4 Ani jumlah=3
  5 Rina jumlah=2
time travel versi 0: 0 | 3 baris
  nama di versi 0: ['Budi', 'Siti', 'Dewi']
berkas _delta_log: ['00000000000000000000.json', '00000000000000000001.json']

Objek lama dibuat dengan version=0, sehingga pembacaan menghimpun hanya aksi add pada commit pertama. Daftar berkas log di baris terakhir memperlihatkan dua commit — pembaca versi 0 mengabaikan 00000000000000000001.json sepenuhnya. Kita akan memanfaatkan sifat ini jauh lebih dalam pada Chapter Time Travel dan Versioning Data.

Schema Evolution dengan schema_mode merge #

Menulis batch yang memuat kolom baru secara langsung akan ditolak Delta Lake — perilaku ini dibedah tuntas di Chapter Schema Enforcement. Dengan schema_mode="merge", kolom baru digabungkan ke schema tabel dan baris lama membaca kolom tersebut sebagai None:

python
import pyarrow as pa
from deltalake import DeltaTable, write_deltalake

BASE = "delta-demo/pesanan"

skema_baru = pa.schema(
    [
        pa.field("id", pa.int64()),
        pa.field("nama", pa.string()),
        pa.field("jumlah", pa.int64()),
        pa.field("kota", pa.string()),
    ]
)
batch = pa.Table.from_arrays(
    [
        pa.array([6], pa.int64()),
        pa.array(["Joko"]),
        pa.array([4], pa.int64()),
        pa.array(["Bandung"]),
    ],
    schema=skema_baru,
)

# tanpa schema_mode, penulisan ini ditolak Delta Lake
try:
    write_deltalake(BASE, batch, mode="append")
    print("penulisan tanpa schema_mode: sukses (?)")
except Exception as galat:
    print("penulisan tanpa schema_mode ditolak:", type(galat).__name__)

# dengan schema_mode="merge", kolom baru digabungkan ke schema tabel
write_deltalake(BASE, batch, mode="append", schema_mode="merge")

dt = DeltaTable(BASE)
isi = dt.to_pyarrow_table()
print("versi:", dt.version(), "|", isi.num_rows, "baris | kolom:", isi.column_names)
for baris in sorted(isi.to_pylist(), key=lambda r: r["id"]):
    print(f"  {baris['id']} {baris['nama']} jumlah={baris['jumlah']} kota={baris['kota']}")

dt_v1 = DeltaTable(BASE, version=1)
print("versi 1 punya kolom:", dt_v1.to_pyarrow_table().column_names)

Output yang Dihasilkan:

text
penulisan tanpa schema_mode ditolak: SchemaMismatchError
versi: 2 | 6 baris | kolom: ['id', 'nama', 'jumlah', 'kota']
  1 Budi jumlah=2 kota=None
  2 Siti jumlah=1 kota=None
  3 Dewi jumlah=5 kota=None
  4 Ani jumlah=3 kota=None
  5 Rina jumlah=2 kota=None
  6 Joko jumlah=4 kota=Bandung
versi 1 punya kolom: ['id', 'nama', 'jumlah']

Perhatikan dua hal dari output di atas. Pertama, penulisan pertama ditolak dengan SchemaMismatchError — penolakan terjadi sebelum satu baris pun tersimpan, sehingga tabel tidak rusak setengah jalan. Kedua, baris terakhir membuktikan bahwa time travel mencakup schema: versi 1 tetap hanya memiliki tiga kolom, sebab kolom kota baru diperkenalkan pada versi 2.

Riwayat Tabel dengan history() #

Setiap commit meninggalkan jejak yang bisa diaudit. Metode history() mengembalikan daftar entri dari yang terbaru, lengkap dengan parameter dan metrik operasi:

python
from deltalake import DeltaTable

BASE = "delta-demo/pesanan"
dt = DeltaTable(BASE)

print("history (terbaru lebih dulu):")
for entri in dt.history():
    metrik = entri.get("operationMetrics") or {}
    print(
        f"  versi {entri['version']} | {entri['operation']}"
        f" | mode={entri['operationParameters'].get('mode')}"
        f" | +{metrik.get('num_added_rows')} baris"
    )

Output yang Dihasilkan:

text
history (terbaru lebih dulu):
  versi 2 | WRITE | mode=Append | +1 baris
  versi 1 | WRITE | mode=Append | +2 baris
  versi 0 | WRITE | mode=ErrorIfExists | +3 baris

Riwayat ini adalah bahan baku audit: siapa menulis apa, kapan, dan berapa baris yang ditambahkan. Nilai mode pada versi 0 adalah ErrorIfExists — mode bawaan write_deltalake yang menolak menimpa tabel yang sudah ada — sedangkan dua versi berikutnya ditulis dengan mode="append".

Ekosistem Delta Lake #

Delta Lake bermula sebagai proyek internal Databricks untuk memberi transaksi ACID pada tabel Parquet di object storage, kemudian dibuka sebagai proyek terbuka dengan versi stabil 3.x. Dukungan engine-nya terbagi dua jalur: Spark memakai Delta secara bawaan melalui format("delta"), sementara engine lain (Trino, Flink, Presto, dan aplikasi Python/Rust/Go) terhubung lewat delta-rs dan konektor resmi. Seluruh contoh chapter ini memakai delta-rs 1.6.6 — rujukan resmi ada di dokumentasi Delta Lake dan dokumentasi delta-rs .

Log JSON per transaksi membuat Delta Lake sangat mudah diinspeksi, tetapi menimbulkan pertanyaan: bagaimana dengan tabel yang berubah baris per baris, bukan batch besar? Untuk kebutuhan itulah Apache Hudi dirancang — topik chapter berikutnya.