Belajar MLOps - Training Pipelines (Kubeflow/Airflow)

#Pendahuluan

Setelah di episode 5-6 eksperimen dan data terkunci, muncul pertanyaan berikutnya: bagaimana cara menjalankan training secara berulang dan terjadwal tanpa orang yang mengklik tombol setiap kali? Data baru masuk tiap minggu — siapa yang menjalankan training ulang? Siapa yang memastikan langkah-langkahnya berurutan dan gagal dengan benar?

Mengapa training pipeline penting? Karena training bukan satu perintah, melainkan rantai langkah: validasi data → preprocessing → training → evaluasi → register model. Menjalankan rantai ini manual adalah sumber kesalahan terbesar di dunia ML — langkah terlewat, urutan salah, atau data tidak sinkron antar langkah. Episode ini mengubah rantai manual menjadi pipeline yang terdefinisi, terjadwal, dan terulang.

#Konsep DAG

Semua orchestrator pipeline memakai pola yang sama: DAG (Directed Acyclic Graph) — grafik berarah tanpa siklus, di mana setiap simpul adalah satu langkah dan panah menyatakan ketergantungan.

DAG menjamin: urutan eksekusi benar, langkah paralel bisa berjalan bersamaan, dan jika satu langkah gagal, dependensinya tidak dijalankan. Semua orchestrator di episode ini — Kubeflow, Airflow, Dagster — adalah engine untuk menjalankan DAG ini.

#Kubeflow Pipelines

Kubeflow adalah platform MLOps native Kubernetes. Kubeflow Pipelines mendefinisikan pipeline sebagai kumpulan component — fungsi Python yang dibungkus jadi container image.

from kfp import dsl
 
@dsl.component(base_image="python:3.12-slim")
def validate_data(data_path: str) -> str:
    import pandas as pd
    df = pd.read_csv(data_path)
    assert df.shape[1] == 12, f"ekspektasi 12 kolom, dapat {df.shape[1]}"
    return "valid"
 
@dsl.component(base_image="python:3.12-slim")
def train_model(valid: str, data_path: str) -> str:
    import mlflow
    # ... training + log_model ...
    return mlflow.get_artifact_uri("model")
 
@dsl.pipeline
def churn_pipeline(data_path: str):
    valid = validate_data(data_path)
    train_model(valid, data_path)

Setiap komponen berjalan di container terpisah di Kubernetes — ini memungkinkan resource berbeda per langkah (training butuh GPU, validasi tidak). Compile dan submit:

kfp compile --py pipeline.py --output pipeline.yaml
kfp pipeline upload --pipeline-name churn pipeline.yaml

Keunggulan Kubeflow: terintegrasi dengan Kubernetes (GPU scheduling dibahas episode 17) dan MLflow. Kelemahannya: kompleks untuk dioperasikan sendiri; banyak tim akhirnya memilih platform terkelola (episode 16).

#Airflow untuk ML

Apache Airflow adalah orchestrator general-purpose paling banyak dipakai — yang juga sangat umum untuk data dan ML pipeline. Berbeda dari Kubeflow yang container-native, Airflow menjalankan task sebagai Python dalam scheduler-nya.

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime
 
def validate_data():
    df = load_data()
    assert df.shape[1] == 12
 
def run_training():
    train_and_log()  # memakai DockerOperator atau PythonOperator
 
def evaluate_and_promote():
    evaluate_and_register()
 
with DAG(
    dag_id="churn_training",
    schedule="0 2 * * 1",  # tiap Senin jam 02:00
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:
    validate = PythonOperator(task_id="validate_data", python_callable=validate_data)
    train = PythonOperator(task_id="train_model", python_callable=run_training)
    evaluate = PythonOperator(task_id="evaluate", python_callable=evaluate_and_promote)
 
    validate >> train >> evaluate

Perhatikan baris schedule="0 2 * * 1" — cron expression yang menjadwalkan pipeline berjalan setiap Senin pukul 02:00. Inilah kekuatan Airflow: scheduling dan retry yang matang. Untuk tugas yang benar-benar butuh container, Airflow punya DockerOperator dan KubernetesPodOperator.

#Dagster: Modern dan Type-aware

Dagster adalah orchestrator generasi baru yang menekankan software-defined assets dan tipe data antar asset. Pendekatan "asset-centric"-nya sangat cocok untuk ML karena setiap dataset/artefak menjadi entitas pertama:

from dagster import asset, materialize
 
@asset
def cleaned_dataset(raw_dataset):
    df = raw_dataset.dropna().drop_duplicates()
    return df
 
@asset
def trained_model(cleaned_dataset):
    model = train(cleaned_dataset)
    return model
 
materialize([cleaned_dataset, trained_model])

Dagster otomatis melacak dependency antar asset dan hanya menjalankan ulang bagian yang berubah — mirip incremental build untuk data.

#Membandingkan Tiga Orchestrator

Tidak ada yang "paling benar" — semuanya memecahkan masalah yang sama (DAG terotomasi). Pilih berdasarkan apa yang tim kalian sudah pakai.

#Best Practice Pipeline Training

• Idempotent step — menjalankan langkah yang sama dua kali menghasilkan output yang sama; penting untuk retry aman.
• Retry dengan backoff — network/GPU transient failure harus di-retry, bukan langsung fail.
• Caching langkah — jika data tidak berubah, jangan jalankan ulang training (Dagster/Kubeflow mendukung ini).
• Alerting — pipeline gagal harus memanggil: email, Slack, atau PagerDuty.
• Resource per step — GPU hanya untuk training, CPU untuk preprocessing — menghemat biaya (episode 22).

#Common Pitfalls

• Pipeline tanpa DAG, hanya script linier → tidak ada retry, tidak ada dependency, tidak ada observabilitas.
• Task tidak idempotent → retry menghasilkan duplikasi atau korupsi data.
• Tanpa alerting → pipeline gagal diam-diam sampai ada yang menyadarinya berminggu-minggu kemudian.
• Menjalankan training besar tanpa resource definition → kacau di shared cluster; definisikan CPU/memori/GPU per step.
• Semua langkah di satu container → tidak ada isolasi failure; pecah jadi komponen.

#Penutup

Pada episode 7 ini, kalian telah mengotomasi alur training menjadi pipeline.

Inti yang harus dibawa pulang:

• Training pipeline = DAG: validasi data → preprocessing → training → evaluasi → register.
• Kubeflow Pipelines: container-native di Kubernetes, cocok tim yang hidup di K8s.
• Airflow: orkestrasi matang dengan scheduling cron dan retry.
• Dagster: asset-centric, hanya menjalankan ulang bagian yang berubah.
• Best practice: idempotency, retry, caching, alerting, dan resource per step.

Di episode 8 selanjutnya kita akan membahas CI/CD untuk ML — GitHub Actions dan GitLab CI yang menjalankan test, validasi data, training, dan quality gates secara otomatis setiap kali kode berubah. Sampai jumpa di episode 8!