Schema Enforcement

Data lake klasik punya pintu masuk yang terbuka lebar: siapa pun bisa menulis berkas Parquet dengan kolom apa pun ke dalam direktori tabel. Sebuah job yang berubah schema-nya tanpa koordinasi akan merusak semua pembaca di hilirnya — query gagal, dashboard berhenti, model ML membaca kolom yang tidak lagi ada. Table format menutup celah ini dengan schema enforcement: penulisan diperiksa terhadap schema tabel sebelum satu baris pun dikomit. Chapter ini menguji mekanisme tersebut secara nyata di Delta Lake, lalu menunjukkan cara mengubah schema dengan aman melalui evolusi, bukan penyerobotan.

Masalah Data yang Tidak Sesuai Schema #

Bayangkan tabel pesanan dengan kolom id, nama, jumlah. Tim A menulis tabel lewat pipeline biasa. Tim B, tanpa memberi tahu, menulis batch baru yang menambahkan kolom kota dan mengubah jumlah menjadi teks. Tanpa enforcement, tiga kegagalan muncul di hilir:

• pembaca engine lama menemukan jumlah kolom yang tidak cocok dan gagal membuka tabel;
• agregasi sum(jumlah) menghasilkan galat tipe atau, lebih buruk, angka yang salah diam-diam;
• laporan yang dibangun sebelumnya berhenti dapat direproduksi karena definisi tabelnya sudah berpindah.

Inti masalahnya: tanpa gerbang tunggal, schema tabel hanyalah kesepakatan lisan antar-tim. Table format mengubahnya menjadi kontrak yang ditegakkan mesin — persis seperti INSERT ke PostgreSQL yang ditolak bila kolomnya tidak cocok.

Enforcement di Table Format vs Ingestion #

Schema bisa diperiksa di dua titik yang berbeda:

Keduanya bukan pilihan salah satu: validasi di ingestion menjaga kualitas isi (nilai, format, domain), sedangkan enforcement di table format menjaga struktur. Kombinasi keduanya adalah praktik yang dianjurkan — bab ini berfokus pada lapisan kedua karena itu yang dimiliki semua table format secara bawaan.

Penolakan oleh Delta Lake, Diuji Nyata #

Blok berikut membuat tabel berisi dua baris, lalu mencoba dua penulisan yang melanggar schema — satu menambah kolom baru, satu mengirim nilai yang tidak dapat menjadi int64. Kedua penolakan dicetak bersama pesan galarinya, diambil dari eksekusi nyata:

import os
import shutil

import pyarrow as pa
from deltalake import DeltaTable, write_deltalake

BASE = "delta-enforce/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, skema=skema):
    return pa.Table.from_arrays(
        [
            pa.array([r[0] for r in rows], skema.field(0).type),
            pa.array([r[1] for r in rows]),
            pa.array([r[2] for r in rows], skema.field(2).type),
        ],
        schema=skema,
    )


write_deltalake(BASE, tabel([(1, "Budi", 2), (2, "Siti", 1)]))
dt = DeltaTable(BASE)
print("tabel dibuat: versi", dt.version(), "|", dt.to_pyarrow_table().num_rows, "baris")

# upaya 1: kolom baru 'kota' tanpa izin evolusi
skema_kota = pa.schema(
    [
        pa.field("id", pa.int64()),
        pa.field("nama", pa.string()),
        pa.field("jumlah", pa.int64()),
        pa.field("kota", pa.string()),
    ]
)
upaya1 = pa.Table.from_arrays(
    [
        pa.array([3], pa.int64()),
        pa.array(["Dewi"]),
        pa.array([5], pa.int64()),
        pa.array(["Bandung"]),
    ],
    schema=skema_kota,
)
try:
    write_deltalake(BASE, upaya1, mode="append")
    print("upaya 1: sukses (?)")
except Exception as galat:
    print("upaya 1 (kolom baru) ditolak:", type(galat).__name__)
    print("  pesan galat:", galat)

# upaya 2: tipe tidak kompatibel, 'lima' tidak bisa menjadi int64
skema_salah = pa.schema(
    [
        pa.field("id", pa.int64()),
        pa.field("nama", pa.string()),
        pa.field("jumlah", pa.string()),
    ]
)
upaya2 = pa.Table.from_arrays(
    [
        pa.array([4], pa.int64()),
        pa.array(["Ani"]),
        pa.array(["lima"]),
    ],
    schema=skema_salah,
)
try:
    write_deltalake(BASE, upaya2, mode="append")
    print("upaya 2: sukses (?)")
except Exception as galat:
    print("upaya 2 (nilai tidak kompatibel) ditolak:", type(galat).__name__)
    print("  pesan galat:", galat)

dt = DeltaTable(BASE)
print("tabel tetap: versi", dt.version(), "|", dt.to_pyarrow_table().num_rows, "baris")

Output yang Dihasilkan:

tabel dibuat: versi 0 | 2 baris
upaya 1 (kolom baru) ditolak: SchemaMismatchError
  pesan galat: Cannot cast schema, number of fields does not match: 4 vs 3
upaya 2 (nilai tidak kompatibel) ditolak: Exception
  pesan galat: Cast error: Cannot cast string 'lima' to value of Int64 type
tabel tetap: versi 0 | 2 baris

Mari kita telaah baris-baris kode di atas:

1. Baris 7-17 (BASE, skema, pembersihan): blok pembuka menghapus direktori lama agar contoh bisa diulang, lalu skema menetapkan tiga kolom; write_deltalake pada baris 31 menetapkan schema tabel pada commit versi 0, dan dua baris Budi serta Siti menjadi isi tabel yang dilindungi.
2. Baris 36-52 (upaya1): batch ini memuat kolom keempat kota — sengaja dibuat dengan schema yang lebih lebar dari tabel.
3. Baris 53-58 (try/except pertama): penulisan ditangkap dan digalatkan; pesan aslinya menyebut jumlah field 4 lawan 3, persis selisih kolom kota.
4. Baris 61-75 (upaya2): jumlah kolom sama, tetapi kolom jumlah dikirim sebagai string "lima"; delta-rs mencoba mengonversinya ke int64 dan gagal pada nilai tersebut.
5. Baris 83-84 (verifikasi akhir): setelah dua penolakan, tabel masih di versi 0 dengan dua baris — penolakan tidak meninggalkan setengah commit.

Evolusi yang Aman #

Penolakan di atas bukan berarti schema tabel beku. Perubahan tetap dimungkinkan selama dinyatakan dengan sadar lewat schema_mode="merge" — cara yang sama dengan Chapter Delta Lake:

import pyarrow as pa
from deltalake import DeltaTable, write_deltalake

BASE = "delta-enforce/pesanan"

skema_kota = pa.schema(
    [
        pa.field("id", pa.int64()),
        pa.field("nama", pa.string()),
        pa.field("jumlah", pa.int64()),
        pa.field("kota", pa.string()),
    ]
)
upaya1 = pa.Table.from_arrays(
    [
        pa.array([3], pa.int64()),
        pa.array(["Dewi"]),
        pa.array([5], pa.int64()),
        pa.array(["Bandung"]),
    ],
    schema=skema_kota,
)

# evolusi aman: kolom baru digabungkan, bukan ditolak
write_deltalake(BASE, upaya1, mode="append", schema_mode="merge")

dt = DeltaTable(BASE)
isi = dt.to_pyarrow_table()
print("setelah evolusi: 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']}")

Output yang Dihasilkan:

setelah evolusi: versi 1 | 3 baris | kolom: ['id', 'nama', 'jumlah', 'kota']
  1 Budi jumlah=2 kota=None
  2 Siti jumlah=1 kota=None
  3 Dewi jumlah=5 kota=Bandung

Kolom baru diterima karena dinyatakan lewat parameter resmi, dan baris lama membaca kota sebagai None alih-alih rusak. Daftar lengkap perilaku schema_mode (termasuk overwrite yang mengganti total schema tabel) tersedia di dokumentasi delta-rs . Klasifikasi perubahan yang aman versus berbahaya:

Perhatikan baris rename pada tabel di atas: Iceberg aman untuk rename karena kolom dikenali lewat field id (dibuktikan pada Chapter Apache Iceberg), sedangkan pada Delta rename sebaiknya dilakukan lewat alat resmi (alter di delta-rs) agar metadata kolom tetap terjaga.

Pola di Iceberg dan Hudi #

Mekanisme gerbangnya serupa di ketiga format. pyiceberg memeriksa kecocokan schema PyArrow terhadap schema tabel sebelum menulis — uji pada Chapter Apache Iceberg memperlihatkan penolakan ketika kolom id yang dideklarasikan required dikirim sebagai kolom nullable, lengkap dengan tabel perbandingan field yang tidak cocok. Hudi memeriksa schema pada saat menulis bersamaan dengan validasi record key, sehingga batch dengan struktur berbeda ditolak sebelum masuk timeline. Filosofinya sama: satu pintu, satu schema, satu kebenaran.

Dengan schema terjaga, pertanyaan berikutnya menyangkut waktu: bagaimana jika kita ingin membuka keadaan tabel seperti kemarin? Itu pekerjaan time travel — topik Chapter Time Travel dan Versioning Data.