Pipeline ML & Praktik Terbaik
Membangun ML Pipeline yang Handal
Pelajari cara membangun pipeline ML yang terstruktur, mencegah data leakage, memastikan reproducibility, dan menghindari kesalahan umum dalam proyek Machine Learning.
🔧 Membangun ML Pipeline
ML Pipeline adalah serangkaian langkah yang terautomatisasi dan terstruktur untuk memproses data hingga menghasilkan model machine learning. Bayangkan pipeline seperti jalur produksi di pabrik — setiap bahan baku (data mentah) melewati serangkaian proses yang terurut sampai menjadi produk jadi (model terlatih).
Tanpa pipeline, proyek ML rentan terhadap kesalahan manual, sulit direproduksi, dan tidak bisa di-deploy dengan mudah. Dengan pipeline, seluruh proses bisa dijalankan ulang hanya dengan satu perintah.
Berikut adalah alur umum sebuah ML Pipeline lengkap:
Data Ingestion
Preprocessing
Feature Eng.
Train/Test Split
Training
Evaluation
Deployment
Setiap kotak di atas adalah sebuah step (tahap) dalam pipeline. Dengan Scikit-learn, setiap step direpresentasikan sebagai objek yang mengimplementasikan metode fit() dan transform().
Scikit-learn menyediakan class Pipeline yang memungkinkan kita menggabungkan beberapa langkah preprocessing dan modeling menjadi satu objek tunggal.
# Import library yang dibutuhkan from sklearn.pipeline import Pipeline from sklearn.preprocessing import StandardScaler, OneHotEncoder from sklearn.impute import SimpleImputer from sklearn.compose import ColumnTransformer from sklearn.ensemble import RandomForestClassifier import pandas as pd # Definisikan kolom numerik dan kategoris numeric_features = ['umur', 'pendapatan', 'lama_bekerja'] categorical_features = ['kota', 'pendidikan'] # Pipeline untuk data numerik numeric_pipeline = Pipeline(steps=[ ('imputer', SimpleImputer(strategy='median')), # isi nilai kosong ('scaler', StandardScaler()) # normalisasi ]) # Pipeline untuk data kategoris categorical_pipeline = Pipeline(steps=[ ('imputer', SimpleImputer(strategy='most_frequent')), # isi nilai kosong ('encoder', OneHotEncoder(handle_unknown='ignore')) # encode ]) # Gabungkan dengan ColumnTransformer preprocessor = ColumnTransformer(transformers=[ ('num', numeric_pipeline, numeric_features), ('cat', categorical_pipeline, categorical_features) ]) # Pipeline LENGKAP: preprocessing + model full_pipeline = Pipeline(steps=[ ('preprocessor', preprocessor), ('classifier', RandomForestClassifier(n_estimators=100, random_state=42)) ]) # Latih sekaligus (fit) dan prediksi (transform) full_pipeline.fit(X_train, y_train) y_pred = full_pipeline.predict(X_test) accuracy = full_pipeline.score(X_test, y_test) print(f"Akurasi: {accuracy:.2f}")
full_pipeline.fit(X_train, y_train), seluruh langkah preprocessing dan training dijalankan secara otomatis dan berurutan. Saat inferensi, predict() juga otomatis menerapkan preprocessing yang sama.
Kadang kita membutuhkan transformasi khusus yang tidak tersedia di Scikit-learn. Kita bisa membuat transformer custom dengan mewarisi BaseEstimator dan TransformerMixin.
from sklearn.base import BaseEstimator, TransformerMixin import numpy as np class LogTransformer(BaseEstimator, TransformerMixin): """Custom transformer untuk log transformation.""" def __init__(self, add_one=True): self.add_one = add_one # tambah 1 agar log(0) tidak error def fit(self, X, y=None): return self # tidak perlu belajar apapun def transform(self, X): if self.add_one: return np.log1p(X) # log(1 + x) return np.log(X) # Gunakan dalam pipeline pipeline_log = Pipeline([ ('log_transform', LogTransformer(add_one=True)), ('scaler', StandardScaler()), ('model', RandomForestClassifier()) ])
Menggunakan Joblib (Direkomendasikan)
import joblib # Simpan pipeline joblib.dump(full_pipeline, 'model_pipeline.pkl') # Muat kembali loaded_pipeline = joblib.load('model_pipeline.pkl') # Langsung prediksi data baru prediksi = loaded_pipeline.predict(data_baru)
Menggunakan Pickle
import pickle # Simpan pipeline with open('pipeline.pkl', 'wb') as f: pickle.dump(full_pipeline, f) # Muat kembali with open('pipeline.pkl', 'rb') as f: loaded = pickle.load(f)
🚫 Mencegah Data Leakage
Data Leakage terjadi ketika informasi dari luar training set “bocor” masuk ke proses pelatihan model, sehingga model terlihat sangat akurat di tahap evaluasi — tetapi gagal total di produksi. Ini adalah salah satu jebakan paling berbahaya dan paling umum dalam ML.
Analogi sederhananya: bayangkan kamu ujian dan sudah tahu kunci jawaban sebelumnya. Nilaimu akan sempurna, tapi kamu belum tentu benar-benar paham materi tersebut.
| Jenis | Deskripsi | Status |
|---|---|---|
| Target Leakage | Fitur yang hanya tersedia SETELAH target diketahui ikut digunakan sebagai input. Contoh: menggunakan “tanggal pelunasan pinjaman” untuk memprediksi apakah seseorang akan gagal bayar. | Berbahaya |
| Train-Test Contamination | Transformasi (seperti scaling atau imputation) di-fit menggunakan seluruh data (termasuk test set), bukan hanya training set. | Umum |
| Temporal Leakage | Untuk data time-series, menggunakan data masa depan untuk memprediksi masa lalu. Misalnya split acak pada data historis harga saham. | Berbahaya |
| Duplicate Leakage | Baris data yang sama muncul di training dan test set sekaligus karena ada duplikasi data yang tidak dibersihkan. | Tersembunyi |
from sklearn.preprocessing import StandardScaler from sklearn.model_selection import train_test_split # ❌ SALAH: scaler di-fit SEBELUM split! scaler = StandardScaler() X_scaled = scaler.fit_transform(X) # pakai seluruh data! X_train, X_test, y_train, y_test = train_test_split( X_scaled, y, test_size=0.2, random_state=42 ) # Model "melihat" statistik test set saat fitting scaler # → Hasil evaluasi TIDAK VALID
# ✅ BENAR: split DULU, baru fit scaler X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42 ) scaler = StandardScaler() # fit HANYA pada training data X_train_scaled = scaler.fit_transform(X_train) # transform test menggunakan statistik training X_test_scaled = scaler.transform(X_test) # Scaler hanya "belajar" dari training data ✓
sklearn.pipeline.Pipeline, Scikit-learn otomatis memastikan bahwa setiap transformer hanya di-fit pada training data saja saat menggunakan cross_validate() atau GridSearchCV().
Data time-series memiliki urutan temporal yang harus dihormati. Jangan pernah menggunakan split acak untuk data ini!
from sklearn.model_selection import TimeSeriesSplit import pandas as pd # Data harga saham (harus terurut berdasarkan waktu) df = df.sort_values('tanggal') # pastikan terurut! # ❌ JANGAN pakai ini untuk time-series: # X_train, X_test = train_test_split(X, test_size=0.2) # shuffle = leakage! # ✅ Gunakan TimeSeriesSplit tscv = TimeSeriesSplit(n_splits=5) for fold, (train_idx, test_idx) in enumerate(tscv.split(X)): X_train, X_test = X[train_idx], X[test_idx] y_train, y_test = y[train_idx], y[test_idx] # Train hanya pakai data HISTORIS # Test selalu ada SETELAH training period print(f"Fold {fold+1}: Train[{train_idx[0]}:{train_idx[-1]}] → Test[{test_idx[0]}:{test_idx[-1]}]")
- Split data SEBELUM melakukan apapun pada data
- Semua transformasi (scaler, imputer, encoder) di-fit hanya pada training set
- Tidak ada fitur yang “tahu” nilai target di masa depan
- Untuk time-series: gunakan TimeSeriesSplit atau split kronologis
- Hapus duplikat sebelum melakukan split
- Gunakan Pipeline Scikit-learn untuk cross-validation
- Verifikasi: jika akurasi terlalu tinggi, investigasi kemungkinan leakage
🔁 Reproducibility dalam ML
Reproducibility berarti kemampuan untuk menjalankan ulang eksperimen ML dan mendapatkan hasil yang identik. Ini krusial karena tanpa reproducibility, kamu tidak tahu apakah peningkatan performa model disebabkan oleh perubahan kode/data, atau hanya keberuntungan acak.
Bayangkan membangun sebuah model yang akurasinya 92%, tapi ketika dijalankan ulang keesokan harinya oleh rekan kerja, hasilnya 85%. Tidak ada yang bisa dipercaya dari model seperti itu.
Machine Learning menggunakan banyak operasi acak: inisialisasi bobot neural network, shuffling data, random split, dll. Dengan menetapkan random seed, kita memaksa “keacakan” ini menjadi konsisten.
import numpy as np import random import os def set_all_seeds(seed=42): """Set semua random seed untuk reproducibility penuh.""" # Python built-in random random.seed(seed) # NumPy np.random.seed(seed) # Hash seed untuk Python (mencegah non-determinisme pada dict) os.environ['PYTHONHASHSEED'] = str(seed) # TensorFlow / Keras (jika digunakan) try: import tensorflow as tf tf.random.set_seed(seed) except ImportError: pass # PyTorch (jika digunakan) try: import torch torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.deterministic = True except ImportError: pass print(f"✅ Semua random seed di-set ke: {seed}") # Panggil di awal setiap notebook/script set_all_seeds(42) # Selalu gunakan random_state di Scikit-learn from sklearn.model_selection import train_test_split X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42 # ← WAJIB! )
Versi berbeda dari library (NumPy, Scikit-learn, XGBoost) bisa menghasilkan angka yang berbeda meskipun kodenya sama. Selalu dokumentasikan environment kamu!
requirements.txt (Minimal)
# Buat requirements.txt pip freeze > requirements.txt # Install di environment baru pip install -r requirements.txt
numpy==1.26.4 pandas==2.2.1 scikit-learn==1.4.2 xgboost==2.0.3 matplotlib==3.8.3 joblib==1.3.2
Conda Environment (Lebih Kuat)
# Export environment lengkap conda env export > environment.yml # Buat environment dari file conda env create -f environment.yml
name: ml-project
channels:
- defaults
- conda-forge
dependencies:
- python=3.11.5
- numpy=1.26.4
- pandas=2.2.1
- scikit-learn=1.4.2
- pip:
- xgboost==2.0.3
MLflow adalah tools open-source untuk mencatat parameter, metrik, dan artefak dari setiap eksperimen ML. Dengan MLflow, kamu bisa membandingkan ratusan eksperimen dan selalu tahu konfigurasi mana yang menghasilkan model terbaik.
import mlflow import mlflow.sklearn from sklearn.ensemble import RandomForestClassifier from sklearn.metrics import accuracy_score, f1_score # Mulai eksperimen mlflow.set_experiment("klasifikasi-churn-pelanggan") with mlflow.start_run(run_name="random-forest-v1"): # Parameter yang digunakan params = { "n_estimators": 100, "max_depth": 5, "random_state": 42 } mlflow.log_params(params) # Latih model model = RandomForestClassifier(**params) model.fit(X_train, y_train) y_pred = model.predict(X_test) # Catat metrik mlflow.log_metric("accuracy", accuracy_score(y_test, y_pred)) mlflow.log_metric("f1_score", f1_score(y_test, y_pred)) # Simpan model sebagai artefak mlflow.sklearn.log_model(model, "model") print("Eksperimen berhasil dicatat di MLflow!") # Jalankan UI: mlflow ui → buka http://localhost:5000
ml-project/ ├── data/ │ ├── raw/ # Data mentah — JANGAN diubah! │ ├── processed/ # Data setelah preprocessing │ └── external/ # Data dari sumber eksternal ├── notebooks/ │ ├── 01_eda.ipynb # Eksplorasi data │ ├── 02_features.ipynb # Feature engineering │ └── 03_modeling.ipynb # Training & evaluasi ├── src/ │ ├── data/ # Script preprocessing │ ├── features/ # Script feature engineering │ ├── models/ # Script training │ └── utils/ # Helper functions ├── models/ # Model tersimpan (.pkl, .joblib) ├── reports/ # Grafik & laporan ├── requirements.txt # Dependensi Python ├── config.yaml # Konfigurasi hyperparameter └── README.md # Dokumentasi proyek
⚠️ Kesalahan Umum dalam ML yang Harus Dihindari
Bahkan data scientist berpengalaman pun masih bisa terjebak pada kesalahan-kesalahan klasik ini. Mengenali dan menghindari kesalahan ini sejak dini akan menghemat berjam-jam debugging dan mencegah model buruk masuk ke produksi.
Banyak pemula langsung menjalankan algoritma ML tanpa memahami data terlebih dahulu. Hasilnya: model yang dibangun di atas fondasi yang salah.
- Langsung fit model ke data mentah
- Tidak cek distribusi fitur
- Tidak cek class imbalance
- Tidak cek outlier ekstrem
- Lakukan EDA menyeluruh terlebih dahulu
- Visualisasikan distribusi semua fitur
- Cek & tangani class imbalance
- Identifikasi dan tangani outlier
Jika dataset kamu memiliki 95% kelas negatif dan 5% kelas positif, model yang selalu memprediksi negatif akan mendapat akurasi 95% — padahal model tersebut sama sekali tidak berguna!
from sklearn.metrics import classification_report, confusion_matrix from sklearn.metrics import roc_auc_score, average_precision_score # ❌ JANGAN hanya gunakan akurasi untuk imbalanced data accuracy = (y_pred == y_test).mean() # bisa menipu! # ✅ Gunakan metrik yang tepat print(classification_report(y_test, y_pred)) # Output: precision, recall, f1-score per kelas # ROC-AUC: bagus untuk imbalanced roc_auc = roc_auc_score(y_test, y_prob) # PR-AUC: lebih baik dari ROC-AUC untuk sangat imbalanced pr_auc = average_precision_score(y_test, y_prob) # Tangani imbalance dengan class_weight from sklearn.ensemble import RandomForestClassifier model = RandomForestClassifier(class_weight='balanced', random_state=42) # Atau gunakan SMOTE untuk oversampling from imblearn.over_sampling import SMOTE X_resampled, y_resampled = SMOTE(random_state=42).fit_resample(X_train, y_train)
Mengevaluasi model dengan satu kali train-test split sangat rentan terhadap keberuntungan. Mungkin saja data yang masuk ke test set kebetulan “mudah” atau “sulit” diprediksi.
from sklearn.model_selection import cross_val_score, StratifiedKFold import numpy as np # ❌ Jangan hanya evaluasi sekali model.fit(X_train, y_train) score = model.score(X_test, y_test) # hasil mungkin tidak representatif # ✅ Gunakan Stratified K-Fold Cross Validation cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42) scores = cross_val_score( pipeline, # gunakan pipeline, bukan model saja! X, y, cv=cv, scoring='f1', n_jobs=-1 # paralel processing ) print(f"F1 Scores per fold: {scores.round(3)}") print(f"Mean F1: {scores.mean():.3f} ± {scores.std():.3f}") # Output: F1 Scores per fold: [0.84 0.86 0.83 0.87 0.85] # Output: Mean F1: 0.850 ± 0.014
Model yang terlalu kompleks akan “hafal” training data tetapi gagal pada data baru. Selalu bandingkan performa training vs validation/test.
# Deteksi overfitting: bandingkan train vs test score train_score = model.score(X_train, y_train) test_score = model.score(X_test, y_test) gap = train_score - test_score print(f"Train Accuracy: {train_score:.3f}") print(f"Test Accuracy: {test_score:.3f}") print(f"Gap : {gap:.3f}") # Diagnosa: if gap > 0.1: print("⚠️ Kemungkinan OVERFITTING — kurangi kompleksitas model") elif test_score < 0.7: print("⚠️ Kemungkinan UNDERFITTING — tingkatkan kompleksitas model") else: print("✅ Model generalisasi dengan baik!") # Solusi overfitting: # - Regularization (L1/L2) # - Kurangi max_depth pada tree # - Tambah data training # - Feature selection # - Dropout (untuk neural network)
Banyak orang langsung membangun model kompleks tanpa baseline. Bagaimana kamu tahu model Random Forest-mu bagus jika tidak dibandingkan dengan model sederhana?
from sklearn.dummy import DummyClassifier # ✅ SELALU buat baseline dulu! # Baseline 1: Prediksi selalu kelas mayoritas dummy_majority = DummyClassifier(strategy='most_frequent') dummy_majority.fit(X_train, y_train) baseline_score = dummy_majority.score(X_test, y_test) print(f"Baseline (majority): {baseline_score:.3f}") # Baseline 2: Model paling sederhana (Logistic Regression) from sklearn.linear_model import LogisticRegression lr = LogisticRegression(random_state=42) lr.fit(X_train, y_train) lr_score = lr.score(X_test, y_test) print(f"Logistic Regression: {lr_score:.3f}") # Model kompleks kamu rf_score = rf_model.score(X_test, y_test) print(f"Random Forest : {rf_score:.3f}") # Jika RF tidak jauh lebih baik dari LR → mungkin tidak perlu kompleksitas ekstra
| # | Kesalahan | Dampak | Solusi |
|---|---|---|---|
| 1 | Langsung modeling tanpa EDA | Tinggi | Selalu eksplorasi data dulu |
| 2 | Hanya pakai akurasi untuk imbalanced | Tinggi | Gunakan F1, ROC-AUC, PR-AUC |
| 3 | Tidak pakai cross-validation | Sedang | StratifiedKFold dengan Pipeline |
| 4 | Tidak cek overfitting | Tinggi | Bandingkan train vs test score |
| 5 | Tidak membuat baseline model | Sedang | Mulai dari DummyClassifier/LR |
| 6 | Data leakage saat preprocessing | Kritis | Gunakan Pipeline + split dulu |
| 7 | Tidak set random seed | Sedang | Set semua seed di awal script |
| 8 | Tidak dokumentasikan eksperimen | Sedang | Gunakan MLflow/W&B |
| 9 | Tidak versioning data & model | Sedang | Gunakan DVC + Git |
| 10 | Langsung deploy tanpa monitoring | Tinggi | Selalu monitor data drift |
✅ Yang Harus Selalu Dilakukan
- Gunakan sklearn Pipeline untuk semua proyek ML
- Split data SEBELUM preprocessing apapun
- Set random seed di awal setiap script
- Dokumentasikan versi library di
requirements.txt - Gunakan cross-validation untuk evaluasi yang valid
- Selalu buat baseline model sebagai pembanding
- Track eksperimen dengan MLflow atau tools sejenis
🚫 Yang Harus Selalu Dihindari
- Jangan fit transformer pada data test
- Jangan gunakan split acak untuk time-series
- Jangan hanya rely pada akurasi untuk imbalanced data
- Jangan langsung ke model kompleks tanpa EDA
- Jangan deploy tanpa memonitor performa di produksi
- Jangan abaikan data leakage meski akurasi bagus