Pipeline Training Otomatis

Pelatihan manual memiliki masa pakai yang pendek: selama satu orang menjalankan script demi script, semuanya berjalan; begitu jadwal berubah atau orangnya cuti, model berhenti diperbarui dan produksi perlahan menua. Pipeline training otomatis mengubah rangkaian langkah manual — siapkan data, latih, evaluasi, daftarkan ke registry — menjadi satu unit yang dijalankan mesin secara terjadwal. Dua sifat wajib menyertainya: idempotency (menjalankan ulang dengan masukan sama tidak menghasilkan efek ganda) dan caching (tahap yang masukannya tidak berubah tidak diulang). Chapter ini membangun pipeline mini yang membuktikan kedua sifat tersebut dengan eksekusi nyata.

Anatomi Pipeline #

Pipeline dideklarasikan sebagai graf berarah tanpa siklus (DAG): setiap node adalah tahap, setiap panah adalah ketergantungan artefak. Bentuk paling umum untuk pelatihan:

flowchart TD A["siapkan_data\n(validasi + bagi split)"] --> B["latih_model"] B --> C["evaluasi\n(metrik validasi)"] C --> D["daftarkan ke registry\nbila lolos gerbang"] D --> E["notifikasi / laporan"]

Kegunaan DAG bukan sekadar gambar: urutan eksekusi dihitung dari ketergantungan, sehingga mesin tahu tahap mana yang boleh berjalan paralel dan tahap mana yang harus menunggu. Ketika satu tahap gagal, hanya tahap itu dan keturunannya yang diulang — bukan seluruh proses dari nol.

Trigger, Idempotency, dan Caching #

Jenis trigger	Contoh	Cocok untuk
terjadwal (schedule)	setiap Senin 02:00	data yang bertambah rutin
berbasis peristiwa (event)	data baru tiba di bucket	data yang datang tak terjadwal
dipicu orang (manual)	tombol “latih ulang sekarang”	perbaikan mendesak
hasil monitoring	drift terdeteksi, PSI > ambang	retraining darurat

Dua sifat teknis membuat pipeline aman dijalankan berulang kali. Idempotency berarti menjalankan tahap dua kali dengan masukan sama meninggalkan efek yang sama dengan sekali — contohnya menulis ulang berkas artefak dengan isi identik, bukan menggandakan entri log. Caching memperluas idempotency: bila hash masukan tahap sama dengan catatan terakhir, tahap tidak dijalankan sama sekali. Hash masukan dibangun dari config tahap ditambah hash artefak tahap sebelumnya — perubahan di hulu otomatis membatalkan cache di hilir.

Contoh Teruji: Pipeline dengan Checkpoint #

Berkas pipeline_mini.py menjalankan tiga tahap berurutan dengan berkas checkpoint cekpoint.json yang mencatat hash masukan tiap tahap. Pipeline dijalankan tiga kali: pertama dari nol, kedua dengan config persis sama, ketiga dengan satu config yang diubah:

python

"""Pipeline training mini: tiga tahap + checkpoint + cache.

Tahapan: siapkan_data -> latih_model -> evaluasi. Setiap tahap
menghitung hash dari masukannya (config + artifact tahap sebelumnya)
dan menyimpannya ke berkas checkpoint. Bila pipeline dijalankan ulang
dengan masukan yang sama, tahap dilewati (cache HIT). Perubahan config
hanya menjalankan ulang tahap yang terdampak.
"""

import hashlib
import json
import os
import shutil

import numpy as np

STORE = "pipeline_store"
CEKPOINT = f"{STORE}/cekpoint.json"


def hash_str(teks):
    return hashlib.sha256(teks.encode()).hexdigest()[:10]


def hash_berkas(nama):
    with open(f"{STORE}/{nama}", "rb") as f:
        return hashlib.sha256(f.read()).hexdigest()[:10]


def baca_cekpoint():
    if os.path.exists(CEKPOINT):
        with open(CEKPOINT, encoding="utf-8") as f:
            return json.load(f)
    return {}


def jalankan_tahap(nama, hash_masukan, fungsi):
    """Eksekusi tahap bila hash masukan berubah; selain itu cache HIT."""
    cek = baca_cekpoint()
    if cek.get(nama) == hash_masukan:
        print(f"  [{nama:<14}] cache HIT  (hash {hash_masukan})")
        return
    print(f"  [{nama:<14}] eksekusi (hash {hash_masukan})")
    fungsi()
    cek[nama] = hash_masukan
    with open(CEKPOINT, "w", encoding="utf-8") as f:
        json.dump(cek, f, indent=2, sort_keys=True)


# ---------- definisi tiga tahap ----------

def tahap_siapkan_data(cfg_data):
    rng = np.random.default_rng(cfg_data["seed"])
    n = cfg_data["n_per_kelas"]
    kelas0 = rng.normal(loc=[-1.0, 0.0], scale=0.7, size=(n, 2))
    kelas1 = rng.normal(loc=[1.0, 0.4], scale=0.7, size=(n, 2))
    X = np.vstack([kelas0, kelas1])
    y = np.array([0] * n + [1] * n)
    urut = rng.permutation(len(y))
    X, y = X[urut], y[urut]
    np.savez(f"{STORE}/data.npz", X=X, y=y)


def tahap_latih(cfg_latih):
    with np.load(f"{STORE}/data.npz") as paket:
        X, y = paket["X"], paket["y"]
    n = len(y)
    w = np.zeros(3)
    for _ in range(cfg_latih["epoch"]):
        skor = np.clip(X @ w[1:] + w[0], -30, 30)
        p = 1.0 / (1.0 + np.exp(-skor))
        galat = p - y
        w[1:] -= cfg_latih["lr"] * (X.T @ galat) / n
        w[0] -= cfg_latih["lr"] * galat.mean()
    np.savez(f"{STORE}/model.npz", bobot=w)


def tahap_evaluasi():
    with np.load(f"{STORE}/data.npz") as paket:
        X, y = paket["X"], paket["y"]
    with np.load(f"{STORE}/model.npz") as paket:
        w = paket["bobot"]
    prediksi = (X @ w[1:] + w[0] >= 0).astype(int)
    akurasi = float((prediksi == y).mean())
    with open(f"{STORE}/metrik.json", "w", encoding="utf-8") as f:
        json.dump({"akurasi_train": round(akurasi, 3)}, f)
    print(f"    -> akurasi_train {akurasi:.3f} dicatat ke metrik.json")


def jalankan_pipeline(cfg_data, cfg_latih, label):
    print(f"== {label} ==")
    hash_data_cfg = hash_str(json.dumps(cfg_data, sort_keys=True))

    jalankan_tahap(
        "siapkan_data", hash_data_cfg,
        lambda: tahap_siapkan_data(cfg_data))
    hash_data = hash_berkas("data.npz")

    jalankan_tahap(
        "latih_model", hash_str(hash_data + json.dumps(cfg_latih,
                                                       sort_keys=True)),
        lambda: tahap_latih(cfg_latih))
    hash_model = hash_berkas("model.npz")

    jalankan_tahap(
        "evaluasi", hash_str(hash_model + hash_data),
        tahap_evaluasi)


def main():
    shutil.rmtree(STORE, ignore_errors=True)
    os.makedirs(STORE)

    cfg_data = {"seed": 42, "n_per_kelas": 120}
    cfg_latih_a = {"lr": 1.0, "epoch": 150}

    jalankan_pipeline(cfg_data, cfg_latih_a, "run 1: dingin (cache kosong)")
    print()
    jalankan_pipeline(cfg_data, cfg_latih_a, "run 2: config sama (semua HIT)")
    print()
    cfg_latih_b = {"lr": 1.0, "epoch": 600}
    jalankan_pipeline(cfg_data, cfg_latih_b,
                      "run 3: epoch 150 -> 600 (hanya tahap terdampak)")

    with open(f"{STORE}/metrik.json", encoding="utf-8") as f:
        print(f"\nmetrik final: {json.dumps(json.load(f))}")
    cek = baca_cekpoint()
    print(f"checkpoint berisi {len(cek)} tahap: "
          f"{', '.join(sorted(cek))}")


if __name__ == "__main__":
    try:
        main()
    finally:
        shutil.rmtree(STORE, ignore_errors=True)

Output yang Dihasilkan:

text
== run 1: dingin (cache kosong) ==
  [siapkan_data  ] eksekusi (hash da49c49e13)
  [latih_model   ] eksekusi (hash 19a63e2936)
  [evaluasi      ] eksekusi (hash 4df77fd946)
    -> akurasi_train 0.921 dicatat ke metrik.json

== run 2: config sama (semua HIT) ==
  [siapkan_data  ] cache HIT  (hash da49c49e13)
  [latih_model   ] cache HIT  (hash 19a63e2936)
  [evaluasi      ] cache HIT  (hash 4df77fd946)

== run 3: epoch 150 -> 600 (hanya tahap terdampak) ==
  [siapkan_data  ] cache HIT  (hash da49c49e13)
  [latih_model   ] eksekusi (hash 4f71446582)
  [evaluasi      ] eksekusi (hash 0193cae6e3)
    -> akurasi_train 0.921 dicatat ke metrik.json

metrik final: {"akurasi_train": 0.921}
checkpoint berisi 3 tahap: evaluasi, latih_model, siapkan_data

Mari kita telaah kode di atas baris per baris:

Baris 21-27 (hash_str, hash_berkas): dua fungsi kecil yang menjadi fondasi caching. Hash berkas dibaca dari bytes artefak di disk, bukan dari nama berkas — nama sama dengan isi berbeda tetap terdeteksi sebagai perubahan.
Baris 37-47 (jalankan_tahap): seluruh kebijakan cache berada di sini. Hash masukan dibandingkan dengan catatan checkpoint; bila sama, tahap dilewati dengan cache HIT. Bila berbeda, fungsi tahap dijalankan lalu hash baru dicatat. Menulis checkpoint setelah eksekusi berhasil (bukan sebelum) membuat tahap yang gagal di tengah otomatis diulang pada percobaan berikutnya.
Baris 52-87 (tiga tahap): tiap tahap menerima artefak tahap sebelumnya melalui berkas di STORE — data.npz diteruskan ke latih_model, model.npz ke evaluasi. Komunikasi lewat berkas inilah yang membuat hash hilir bisa mendeteksi perubahan di hulu.
Baris 90-107 (jalankan_pipeline): hash masukan tiap tahap dirangkai secara eksplisit — tahap latih_model menggabungkan hash berkas data dengan config latih, tahap evaluasi menggabungkan hash model dan hash data. Rantai hash inilah DAG dalam bentuk yang bisa dihitung.
Baris 117-123 (tiga run): run 1 dieksekusi penuh; run 2 dengan config sama menghasilkan tiga cache HIT; run 3 mengubah epoch sehingga hanya latih_model dan evaluasi yang berjalan ulang sementara siapkan_data tetap cache HIT — tepat perilaku yang dijanjikan caching berbasis hash.

Pada run 3, perhatikan bahwa hash siapkan_data tidak berubah (da49c49e13) karena config data memang sama, tetapi hash latih_model berubah karena config epoch baru — dan hash evaluasi ikut berubah karena model.npz yang dihasilkan berbeda. Perubahan menjalar ke hilir secara otomatis; tidak ada cache basi yang lolos.

Orkestrasi di Dunia Nyata #

Untuk pipeline produksi, dua nama mendominasi. Apache Airflow mendefinisikan tahap sebagai operator Python dengan penjadwal bawaan, tampilan status per run, dan percobaan ulang otomatis saat gagal. Kubeflow Pipelines mengeksekusi tiap tahap sebagai container terpisah — pilihan cocok ketika tahap menuntut lingkungan berbeda (preprocessing di pandas, training di GPU). Keduanya menyelesaikan hal yang sama dengan jalankan_tahap kita: menjalankan DAG, menyimpan artefak, dan menawarkan jalur ulang yang murah.

Pipeline mengotomatiskan produksi kandidat model — tetapi setiap perubahan kode pipeline sendiri juga perlu diuji sebelum dipercaya berjalan sendiri. Chapter berikutnya, Continuous Integration , membawa disiplin test perangkat lunak ke dalam proyek machine learning.