Data Lineage dan Provenance

Angka penjualan di dashboard TokoKita tiba-tiba berubah satu hari setelahnya, dan tim bertanya: dari sumber mana angka itu dihitung, versi kode apa yang mengolahnya, dan apakah berkas hasilnya masih hasil yang sama dengan saat laporan dibuat? Chapter Data Lineage di section Data Quality sudah menjawab separuh pertanyaan — peta alur data dari sumber ke tabel akhir. Bab ini melengkapi separuh lainnya dari sisi tata kelola: provenance, yaitu rekaman lengkap bagaimana sebuah artefak data dihasilkan — input apa dengan hash berapa, kode versi berapa, parameter apa — sehingga setiap angka bisa direproduksi dan setiap perubahan diam-diam terdeteksi. Kombinasi keduanya adalah syarat audit yang kuat: lineage menunjukkan jalurnya, provenance membuktikan isinya.

Lineage vs Provenance #

Keduanya sering dipakai bergantian, padahal fokusnya berbeda. Lineage memetakan hubungan antar node — tabel, kolom, job — dan menjawab pertanyaan alur. Provenance merekam riwayat satu artefak secara mendalam dan menjawab pertanyaan reproduksi:

Kita sudah menempatkan dua fondasi untuk bab ini: tabel run_metadata di Chapter Pipeline Versioning mencatat versi kode per run, dan versioning data di lakehouse (Chapter Time Travel dan Versioning Data) menjaga riwayat isi tabel. Provenance menyatukan keduanya menjadi satu rekaman lengkap per run.

Implementasi: Merekam Provenance Run #

Pipeline contoh kita membaca sumber.jsonl (tiga pesanan), menghitung total penjualan plus pajak, lalu menulis ringkasan.json. Setiap kali pipeline dijalankan, program juga menulis provenance.json: identitas run, versi kode hasil pola git describe --tags, parameter, hash sha256 input beserta jumlah barisnya, dan hash output. Versi kode di sini berasal dari konvensi tag yang sudah kita bangun di Chapter Pipeline Versioning, sehingga rekaman provenance dan riwayat git selalu merujuk identitas yang sama:

"""Rekam provenance run pipeline: input+sha256, versi kode, parameter, output+sha256."""
import hashlib
import json
from pathlib import Path

DASAR = Path(__file__).parent
SUMBER = DASAR / "sumber.jsonl"
OUTPUT = DASAR / "ringkasan.json"
PROVENANCE = DASAR / "provenance.json"

# Hasil `git describe --tags` saat rilis - pola dari chapter Pipeline Versioning.
KODE_VERSI = "v1.2.0"
PARAMETER = {"pajak_persen": 11.0, "mata_uang": "IDR"}


def sha256(berkas):
    return hashlib.sha256(berkas.read_bytes()).hexdigest()


def jumlah_baris(berkas):
    return len([b for b in berkas.read_text().splitlines() if b.strip()])


def transform():
    """Hitung total penjualan lalu tulis ringkasan output."""
    total = 0
    for baris in SUMBER.read_text().splitlines():
        data = json.loads(baris)
        total += data["qty"] * data["harga"]
    pajak = round(total * PARAMETER["pajak_persen"] / 100)
    OUTPUT.write_text(json.dumps(
        {"total_penjualan": total, "pajak": pajak}, indent=2) + "\n")


def jalankan():
    """Jalankan pipeline lalu simpan rekam provenance run ini."""
    transform()
    rekam = {
        "run_id": "run-20260801-001",
        "waktu": "2026-08-01T02:15:00+07:00",
        "kode_versi": KODE_VERSI,
        "parameter": PARAMETER,
        "input": {
            "path": SUMBER.name, "sha256": sha256(SUMBER),
            "baris": jumlah_baris(SUMBER),
        },
        "output": {"path": OUTPUT.name, "sha256": sha256(OUTPUT)},
    }
    PROVENANCE.write_text(json.dumps(rekam, indent=2) + "\n")
    print("Provenance tersimpan ke provenance.json:")
    print(json.dumps(rekam, indent=2))


def verifikasi():
    """Hitung ulang sha256 output; bandingkan dengan yang tercatat saat run."""
    rekam = json.loads(PROVENANCE.read_text())
    aktual = sha256(OUTPUT)
    if aktual == rekam["output"]["sha256"]:
        return (f"OK: {rekam['output']['path']} tidak berubah sejak "
                f"run {rekam['run_id']} (kode {rekam['kode_versi']})")
    return (f"GAGAL: hash output berubah sejak run {rekam['run_id']}!\n"
            f"  tercatat: {rekam['output']['sha256'][:16]}...\n"
            f"  aktual  : {aktual[:16]}...")


if __name__ == "__main__":
    jalankan()
    print("Verifikasi :", verifikasi())
    # Skenario: proses lain menulis ulang output tanpa mencatat provenance baru.
    data = json.loads(OUTPUT.read_text())
    data["total_penjualan"] += 999999
    OUTPUT.write_text(json.dumps(data, indent=2) + "\n")
    print("Setelah output dimodifikasi pihak lain:")
    print("Verifikasi :", verifikasi())

Output yang Dihasilkan:

Provenance tersimpan ke provenance.json:
{
  "run_id": "run-20260801-001",
  "waktu": "2026-08-01T02:15:00+07:00",
  "kode_versi": "v1.2.0",
  "parameter": {
    "pajak_persen": 11.0,
    "mata_uang": "IDR"
  },
  "input": {
    "path": "sumber.jsonl",
    "sha256": "5cc45d80aec75b3b43dc622cbc3d0abb4a7a2cfccb7582b0f0290317f9df8acc",
    "baris": 3
  },
  "output": {
    "path": "ringkasan.json",
    "sha256": "ef1510c01ccd623841e88da56989beee6ada0b93df2773bff0ceb3150f44ec07"
  }
}
Verifikasi : OK: ringkasan.json tidak berubah sejak run run-20260801-001 (kode v1.2.0)
Setelah output dimodifikasi pihak lain:
Verifikasi : GAGAL: hash output berubah sejak run run-20260801-001!
  tercatat: ef1510c01ccd6238...
  aktual  : f3b36dd743b455f0...

Mari kita telaah baris-baris kode di atas:

1. Baris 16-17 (sha256): menghitung hash atas byte berkas secara penuh — hash sha256 inilah yang membuat “berkas yang sama” berarti identik bit demi bit, bukan sekadar nama sama.
2. Baris 38-48 (rekam): rekaman provenance dibangun setelah transform() selesai, sehingga hash output mengunci hasil yang benar-benar tertulis — termasuk identitas input, parameter, dan versi kode dalam satu objek.
3. Baris 54-63 (verifikasi): membaca provenance, menghitung ulang hash output saat ini, lalu membandingkan. Cocok berarti output masih persis hasil run yang tercatat; tidak cocok berarti ada proses lain yang menyentuh berkas tanpa mencatat run baru.
4. Baris 69-74 (skenario modifikasi): total_penjualan ditambah sembilan ratus ribu oleh proses lain tanpa provenance baru. Verifikasi berikutnya langsung melaporkan hash tercatat berbeda dari hash aktual — perubahan diam-diam gagal lolos.

Bagaimana rekaman ini berhubungan dengan peta alur? Satu edge lineage sederhana bisa digambar langsung dari provenance yang kita rekam:

Node provenance.json menghubungkan job dengan hash kedua ujungnya: ketika edge semacam ini terkumpul dari banyak run, grafnya menjadi lineage otomatis — pola yang diadopsi sebagai standar industri oleh OpenLineage . Model konseptual untuk merekam asal-usul artefak secara umum sudah lama distandarkan W3C di spesifikasi PROV , dan rekaman JSON kita adalah bentuk sederhana dari entity–activity–agent pada standar tersebut.

Rekaman ini juga membuka jalur replay: untuk membuktikan sebuah angka, kita men-checkout kode versi v1.2.0 dari git, memastikan input masih ber-hash sama, menjalankan ulang pipeline dengan parameter yang terekam, lalu membandingkan hash output baru dengan hash di provenance. Bila kedua hash cocok, angka terbukti reproduksibel; bila tidak, ada nondeterminisme yang perlu dicari — misalnya timestamp eksekusi ikut tertulis ke output.

Rekaman provenance menjawab “bagaimana data ini dihasilkan”, dan katalog menjawab “aset apa ini dan siapa yang mengelolanya”. Pertanyaan penentu berikutnya: ketika aset bermasalah pada jam dua pagi, siapa orang yang bertanggung jawab dan berani dihubungi? Itu wilayah Chapter Data Ownership.