Belajar ML Engineer - Data Pipelines untuk ML

#Pendahuluan

Setelah di episode 16 model kalian menjawab dari basis pengetahuan, sekarang kita mundur satu lapisan ke bagian yang paling tidak glamor tapi paling menentukan: data pipeline. Semua yang sudah kalian bangun — feature, training, monitoring — bergantung pada satu pertanyaan: apakah data yang masuk ke sistem benar, segar, dan konsisten?

MLE sering menemukan dirinya di persimpangan: model bagus, tapi data pipeline berantakan sehingga training selalu pakai data telat atau bermasalah. Episode ini membangun pipeline yang solid: cara menyerap data dari berbagai sumber, memprosesnya secara batch maupun streaming, mengatur orkestrasi, dan — yang paling penting untuk MLE — feature store yang menyatukan feature untuk training dan serving.

#Ingestion: Mengambil Data dari Sumber

Data datang dari banyak tempat: database transaksional (PostgreSQL), log aplikasi (Kafka), file (S3), API eksternal. Pipeline ingestion menggerakkan data ke tempat yang bisa diproses:

Prinsip yang wajib: data mentah bersifat immutabel (sudah kita bahas di episode 8). Pipeline ingestion menyimpan raw ke object storage (S3/GCS) sebagai source of truth, dan semua proses turunan membaca dari sana — bukan mengubahnya.

#Processing: Batch vs Streaming

Dua mode pemrosesan dengan trade-off latency vs kompleksitas:

Prinsip yang mencegah kekacauan: jangan memproses streaming kalau batch sudah cukup. Streaming menambah biaya operasional besar; gunakan hanya saat data memang harus sampai dalam hitungan detik (misal fraud scoring real-time).

def build_training_dataset(period_start: str, period_end: str) -> pd.DataFrame:
    raw = read_raw_events(period_start, period_end)   # dari object storage
    cleaned = clean(raw)                              # dedup, tipe data, missing
    aggregated = aggregate_features(cleaned)          # feature engineering
    aggregated.to_parquet(f"datasets/events_{period_start}_{period_end}.parquet")
    return aggregated

#Orkestrasi: Mengatur Urutan dan Waktu

Saat pipeline punya banyak tahap dengan dependensi (ingest → clean → aggregate → train), kalian butuh orchestrator — bukan bash cron raksasa. Dua pilihan utama:

• Apache Airflow: DAG berbasis Python, scheduler, retry, UI — standar lama yang tetap dominan.
• Prefect / Dagster: lebih modern, dynamic scheduling, testing yang lebih mudah.

Konsep inti orkestrasi yang harus dipahami:

• DAG: tahap + dependensi; kalian tidak pernah menjalankan tahap sebelum prasyaratnya.
• Retry dengan backoff: transient failure (API down) tidak menghentikan seluruh pipeline.
• Idempotency: menjalankan ulang pipeline harus menghasilkan output yang sama — atau pipeline akan menumpuk duplikat.
• Observability: setiap run tercatat, bisa di-inspect, dan punya log.

from prefect import flow, task
 
@task
def ingest():
    ...
 
@task
def clean(df):
    ...
 
@flow
def ml_data_pipeline():
    raw = ingest()
    df = clean(raw)
    features = build_features(df)
    features.to_parquet("datasets/train_latest.parquet")

#Feature Store: Menyatukan Training dan Serving

Ini bagian paling penting bagi MLE. Feature store adalah sistem terpusat yang menyimpan feature sehingga:

• Training mengambil nilai feature untuk data historis.
• Serving mengambil nilai feature yang sama untuk request live.
• Keduanya memakai definisi feature yang sama — bukan dua implementasi.

Tanpa feature store:
  training  → hitung feature sendiri (script A)
  serving   → hitung feature sendiri lagi (script B, berbeda!)
  → training-serving skew (masalah besar di episode 4)
 
Dengan feature store:
  training  ─┐
              ├──► Feature Store (definisi tunggal)
  serving   ─┘        │
                      ├──► train model
                      └──► serve model

Fitur kunci yang diharapkan dari feature store: definisi feature terpusat, point-in-time correctness (mengambil nilai feature "pada waktu yang benar" untuk tiap baris, menghindari data leakage dari masa depan), dan online serving dengan latency rendah.

Pilihan tooling: Feast (open source, self-hosted), Featureform, atau platform cloud (SageMaker Feature Store, Vertex AI Feature Store). Mulai dari konsep dulu — sering kali solusi sederhana (tabel terpusat + query time) sudah cukup sebelum menambah platform.

#Memastikan Kualitas Data di Pipeline

Pipeline tanpa checkpoint kualitas akan mengalirkan sampah ke training. Masukkan validasi otomatis di setiap tahap:

def validate(df: pd.DataFrame) -> None:
    assert df.notna().mean().min() > 0.95, "missing rate terlalu tinggi"
    assert df["event_count"].min() >= 0, "nilai negatif tidak valid"
    schema = {"user_id": "int64", "event_count": "int64"}
    assert df.dtypes.astype(str).to_dict() == schema, "schema berubah"

Gabungkan dengan data validation tool seperti Great Expectations atau pandas-based assert. Pipeline yang meloloskan data bermasalah adalah hutang teknis yang dibayar berkali-kali di tahap model.

#Common Pitfalls

#Penutup

Pada episode 17 ini, kalian telah membangun fondasi data:

• Ingestion dari DB, log, file, dan API — raw data immutabel di object storage.
• Processing batch vs streaming; pilih batch kecuali benar-benar butuh real-time.
• Orkestrasi dengan DAG (Airflow/Prefect): retry, idempotency, observability.
• Feature store menyatukan training & serving — anti training-serving skew.
• Gate kualitas data yang gagal dengan keras, bukan diam-diam.

Di episode 18 selanjutnya kita akan membahas ML Security — adversarial robustness, prompt injection, dan data poisoning: ancaman yang menargetkan model, pipeline data, dan cara pertahanannya. Sampai jumpa di episode 18!