Tim finance toko online kita menemukan bahwa angka laporan Maret berubah setelah bug pipeline diperbaiki dan data diproses ulang, lalu seorang auditor menanyakan hal yang sederhana namun menusuk: versi kode mana yang menghasilkan angka lama, dan versi mana yang menghasilkan angka baru? Tanpa catatan versi, pertanyaan tersebut tidak bisa dijawab — reproduksi angka menjadi mustahil dan setiap perbandingan historis kehilangan dasar. Versioning menutup celah tersebut dengan mencatat dua versi sekaligus: versi kode yang menjalankan pipeline dan versi skema dari output yang dihasilkan.

Dua Versi yang Wajib Dicatat #

Versi kode melacak isi pipeline saat run terjadi; versi output melacak struktur tabel hasil. Keduanya berubah dengan ritme berbeda — kode setiap perbaikan bug, skema setiap kolom ditambah atau diubah — sehingga keduanya wajib dicatat terpisah. Run yang memakai kode v1.1.0 bisa saja masih menulis skema v1, dan justru kombinasi keduanya yang menjelaskan asal-usul sebuah angka.

AspekVersi kodeVersi skema output
Yang dilacakisi kode pipeline saat runstruktur tabel hasil
Penandatag git, misalnya v1.0.0nama tabel versi, misalnya pesanan_v1
Berubah saatlogika transform diperbaikikolom ditambah, tipe diubah
Menjawab pertanyaankode apa yang membuat data ini?struktur apa yang boleh dibaca konsumen?

Git Tag dan Tabel Metadata Run #

Tag git menandai titik rilis kode secara tidak bisa diubah, dan perintah git describe --tags mengembalikan tag terdekat dari commit yang sedang berjalan — misalnya v1.0.0, atau v1.0.0-3-g1a2b3c bila ada tiga commit setelah tag. Pipeline memanggil perintah tersebut saat runtime, lalu menuliskan hasilnya ke tabel run_metadata bersama versi skema output, tabel target, dan jumlah baris. Dengan tabel ini, setiap tabel output bisa ditelusuri balik: siapa yang menulisnya, dengan kode versi berapa, dan menghasilkan berapa baris. Konvensi penomoran yang lazim: naikkan angka major untuk perubahan yang memaksa konsumen menyesuaikan diri, naikkan angka minor untuk perubahan yang tetap kompatibel.

Migrasi Paralel: Skema v1 dan v2 #

Ketika struktur output harus berubah, jangan mengubah tabel lama secara langsung — konsumen yang sudah berjalan akan patah begitu perubahan dideploy. Pola yang aman adalah paralel: buat tabel baru pesanan_v2, biarkan pipeline menulis ke kedua tabel selama masa migrasi, pindahkan konsumen satu per satu ke v2, lalu pensiunkan v1 setelah tidak ada yang membacanya. Pendekatan yang sama berlaku untuk berkas: tulis ke path ber-versi, bukan menimpa path yang sama.

Implementasi: Mencatat Versi Kode ke Metadata Run #

Kita mulai dari pipeline versi pertama. Direktori proyek dijadikan repositori git, kode diberi tag v1.0.0, lalu pipeline dijalankan — identitas kode dibaca dari tag dan dicatat ke tabel run_metadata.

bash 
git init -b main
git config user.name "Budi"
git config user.email "[email protected]"
echo "*.db" > .gitignore
git add .gitignore pipeline.py sumber.jsonl
git commit -m "pipeline pesanan harian, skema output v1"
git tag -a v1.0.0 -m "skema output v1"
python 
"""Pipeline pesanan harian: catat versi kode dan versi skema output ke metadata run."""
import json
import sqlite3
import subprocess
from pathlib import Path

SCHEMA_OUTPUT = "v1"


def versi_kode():
    """Baca identitas kode dari tag git, misalnya v1.0.0."""
    return subprocess.run(
        ["git", "describe", "--tags"], capture_output=True, text=True, check=True
    ).stdout.strip()


def transform(baris):
    data = json.loads(baris)
    total = data["qty"] * data["harga"] - (data.get("diskon") or 0)
    return (data["pesanan_id"], data["waktu"], data["pelanggan"],
            data["produk"], data["qty"], data["harga"], total)


def jalankan():
    koneksi = sqlite3.connect("toko_versi.db")
    koneksi.executescript("""
    CREATE TABLE IF NOT EXISTS run_metadata (
        run_id INTEGER PRIMARY KEY AUTOINCREMENT,
        kode_versi TEXT NOT NULL, schema_output TEXT NOT NULL,
        tabel_target TEXT NOT NULL, jumlah_baris INTEGER NOT NULL);
    CREATE TABLE IF NOT EXISTS pesanan_v1 (
        pesanan_id INTEGER PRIMARY KEY, waktu TEXT NOT NULL,
        pelanggan TEXT NOT NULL, produk TEXT NOT NULL,
        qty INTEGER NOT NULL, harga INTEGER NOT NULL, total INTEGER NOT NULL);
    """)
    jumlah = 0
    for baris in Path("sumber.jsonl").read_text().splitlines():
        koneksi.execute(
            "INSERT OR REPLACE INTO pesanan_v1 VALUES (?, ?, ?, ?, ?, ?, ?)",
            transform(baris))
        jumlah += 1
    koneksi.execute(
        """INSERT INTO run_metadata (kode_versi, schema_output, tabel_target, jumlah_baris)
           VALUES (?, ?, ?, ?)""",
        (versi_kode(), SCHEMA_OUTPUT, "pesanan_v1", jumlah))
    koneksi.commit()
    print(f"run selesai: kode {versi_kode()} -> skema output {SCHEMA_OUTPUT}, "
          f"{jumlah} baris ke pesanan_v1")
    koneksi.close()


if __name__ == "__main__":
    jalankan()

Output yang Dihasilkan:

text 
run selesai: kode v1.0.0 -> skema output v1, 8 baris ke pesanan_v1

Mari kita telaah baris-baris kode di atas:

  1. Baris 7 (SCHEMA_OUTPUT = "v1"): versi skema output dinyatakan eksplisit sebagai konstanta, bukan diketebak dari struktur tabel — satu titik kebenaran yang ikut tercatat ke metadata.
  2. Baris 10-14 (versi_kode): menjalankan git describe --tags lewat subprocess.run dan mengembalikan identitas kode, misalnya v1.0.0. Argumen check=True membuat pipeline gagal keras bila direktori kerja bukan repositori git — lebih baik gagal daripada mencatat versi kosong.
  3. Baris 17-21 (transform): transformasi v1 menghasilkan tujuh kolom tanpa pajak; versi inilah yang dikunci tag v1.0.0.
  4. Baris 27-34 (dua tabel): run_metadata menyimpan riwayat eksekusi, sementara pesanan_v1 adalah output berversi — nama tabelnya sudah menyandang versi skema.
  5. Baris 42-45 (insert metadata): setelah memuat, pipeline menuliskan versi kode hasil versi_kode(), versi skema, tabel target, dan jumlah baris. Di produksi, baris ini biasanya diperkaya waktu mulai, durasi, dan status run.

Menjalankan v2 Berdampingan dengan v1 #

Kebutuhan baru datang: tim finance meminta kolom pajak sebesar 11% dari total. Alih-alih mengubah pesanan_v1, transform naik ke v2 dan hasilnya ditulis ke tabel baru pesanan_v2 yang memiliki kolom tambahan tersebut. Perubahan pada fungsi transform tampak seperti ini:

python Perubahan transform menuju v2 
def transform(baris):
    data = json.loads(baris)
    total = data["qty"] * data["harga"] - (data.get("diskon") or 0)
    pajak = round(total * 0.11)
    return (data["pesanan_id"], data["waktu"], data["pelanggan"],
            data["produk"], data["qty"], data["harga"], total, pajak)

Selain fungsi tersebut, bagian jalankan menyesuaikan: tabel pesanan_v2 dibuat dengan kolom pajak, pernyataan INSERT menerima delapan nilai, dan konstanta SCHEMA_OUTPUT bernilai "v2". Perubahan lalu di-commit dan diberi tag baru:

bash 
git add pipeline.py
git commit -m "tambah kolom pajak, skema output v2 (tabel baru pesanan_v2)"
git tag -a v1.1.0 -m "skema output v2"

Output yang Dihasilkan:

text 
run selesai: kode v1.1.0 -> skema output v2, 8 baris ke pesanan_v2

Skrip kecil berikut membaca kembali riwayat yang tersimpan, lalu membandingkan isi kedua tabel — persis pertanyaan auditor di awal bab ini.

python 
"""Periksa riwayat run: versi kode mana yang menghasilkan tabel output."""
import sqlite3

koneksi = sqlite3.connect("toko_versi.db")

print("=== Metadata run (riwayat eksekusi) ===")
for run_id, kode, skema, tabel, jumlah in koneksi.execute(
    "SELECT run_id, kode_versi, schema_output, tabel_target, jumlah_baris"
    " FROM run_metadata ORDER BY run_id"
):
    print(f"  run {run_id}: kode {kode}, skema {skema}, {tabel}, {jumlah} baris")

print("=== Konsumen lama: masih membaca pesanan_v1 ===")
for pesanan_id, total in koneksi.execute(
    "SELECT pesanan_id, total FROM pesanan_v1 WHERE pesanan_id IN (1, 5)"
):
    print(f"  pesanan {pesanan_id}: total {total}")

print("=== Konsumen baru: pindah ke pesanan_v2 dengan kolom pajak ===")
for pesanan_id, total, pajak in koneksi.execute(
    "SELECT pesanan_id, total, pajak FROM pesanan_v2 WHERE pesanan_id IN (1, 5)"
):
    print(f"  pesanan {pesanan_id}: total {total}, pajak {pajak}")

koneksi.close()

Output yang Dihasilkan:

text 
=== Metadata run (riwayat eksekusi) ===
  run 1: kode v1.0.0, skema v1, pesanan_v1, 8 baris
  run 2: kode v1.1.0, skema v2, pesanan_v2, 8 baris
=== Konsumen lama: masih membaca pesanan_v1 ===
  pesanan 1: total 460000
  pesanan 5: total 200000
=== Konsumen baru: pindah ke pesanan_v2 dengan kolom pajak ===
  pesanan 1: total 460000, pajak 50600
  pesanan 5: total 200000, pajak 22000

Riwayat run kini menjawab pertanyaan auditor dengan data, bukan ingatan: run 1 oleh kode v1.0.0 menulis 8 baris ke pesanan_v1, run 2 oleh kode v1.1.0 menulis 8 baris ke pesanan_v2. Konsumen lama tetap membaca pesanan_v1 tanpa terganggu, konsumen baru mendapatkan kolom pajak dari pesanan_v2, dan kapan pun sebuah angka dipertanyakan, kita tahu persis kombinasi kode dan skema yang menghasilkannya.

Setiap run kini tercatat identitasnya. Catatan tersebut juga menjadi fondasi saat satu kode pipeline ingin dipakai ulang untuk banyak sumber dan banyak toko sekaligus — pokok bahasan Chapter Pipeline Reusability.