Membangun Pipeline Pemrosesan

Rencana sudah berjalan, dan pola yang sama mulai berulang tiap pekan: data mentah baru tiba, lalu kita membersihkan baris ganda, menambal nilai kosong, membentuk kolom fitur, dan menyimpan hasilnya. Seminggu pertama pekerjaan itu masih masuk akal dikerjakan dengan menyalin sel kode lama; pada bulan kedua, salin-tempel menjadi beban — satu perbaikan kecil harus disalin ke banyak tempat, dan satu tempat yang terlewat merusak hasil tanpa memberi tahu. Pipeline pemrosesan menuntas masalah itu: seluruh langkah dirangkai menjadi satu rangkaian fungsi yang dijalankan dari atas ke bawah, menghasilkan keluaran yang sama setiap kali diberi masukan yang sama. Pada chapter ini kita membangun pipeline dari fungsi murni, lalu melihat versi scikit-learn untuk saat model ikut terlibat.

Fungsi Murni sebagai Blok Bangunan #

Blok bangunan pipeline yang sehat adalah fungsi murni (pure function): keluarannya hanya bergantung pada masukannya, dan fungsi tersebut tidak mengubah keadaan apa pun di luar dirinya. Fungsi bersihkan(data) yang mengembalikan data bersih tanpa menyentuh data asli adalah murni; kode yang menimpa variabel global atau memodifikasi DataFrame yang sama di tempatnya bukan. Kemurnian ini yang membuat pipeline aman: setiap tahap bisa diuji sendirian, dan hasil antara bisa diperiksa tanpa takut tahap berikutnya mengubahnya diam-diam.

Dua kebiasaan kecil menjaga kemurnian itu. Pertama, fungsi mengembalikan objek baru — drop_duplicates() dan copy() sudah berperilaku begitu, dan menambah copy() pada hasilnya membuat niatnya eksplisit. Kedua, efek samping yang tidak terhindarkan — menulis berkas, mencetak laporan — dipisahkan ke tahap paling akhir, sehingga bagian tengah pipeline tetap murni dan mudah diuji.

Pipeline Murni Python: Muat, Bersihkan, Fitur, Simpan #

Kita bangun pipeline lengkap untuk data transaksi Toko Kopi Kita dengan empat tahap: muat, bersihkan, tambah fitur, simpan. Setiap tahap adalah fungsi yang menerima DataFrame dan mengembalikan DataFrame (atau laporan, untuk tahap simpan), dan setiap tahap mencetak satu baris ringkasan supaya kita bisa melihat bentang datanya berubah di sepanjang jalan.

import os

import pandas as pd

HARI = ["Senin", "Selasa", "Rabu", "Kamis", "Jumat", "Sabtu", "Minggu"]

def muat() -> pd.DataFrame:
    "Tahap 1: sediakan data mentah dari sumber."
    mentah = pd.DataFrame({
        "tanggal": ["2026-01-05", "2026-01-05", "2026-01-06", "2026-01-05",
                    "2026-01-06", "2026-01-07", "2026-01-07", "2026-01-08"],
        "pelanggan": ["Budi", "Siti", "Dewi", "Siti",
                      "Budi", "Dewi", None, "Agus"],
        "total": [24000, 18000, 28000, 18000, 24000, 28000, 22000, 32000],
    })
    print(f"[muat]      {len(mentah)} baris mentah")
    return mentah

def bersihkan(mentah: pd.DataFrame) -> pd.DataFrame:
    "Tahap 2: buang baris ganda dan beri label pelanggan nonmember."
    bersih = mentah.drop_duplicates().copy()
    bersih["pelanggan"] = bersih["pelanggan"].fillna("nonmember")
    print(f"[bersihkan] {len(bersih)} baris setelah baris ganda dibuang")
    return bersih

def tambah_fitur(bersih: pd.DataFrame) -> pd.DataFrame:
    "Tahap 3: turunkan fitur baru dari kolom yang sudah ada."
    berfitur = bersih.copy()
    tanggal = pd.to_datetime(berfitur["tanggal"])
    berfitur["hari"] = tanggal.dt.dayofweek.map(dict(enumerate(HARI)))
    berfitur["total_ribu"] = berfitur["total"] // 1000
    print(f"[fitur]     {len(berfitur)} baris, {berfitur.shape[1]} kolom")
    return berfitur

def simpan(berfitur: pd.DataFrame, nama_berkas: str) -> None:
    "Tahap 4: satu-satunya efek samping - tulis berkas lalu verifikasi."
    berfitur.to_csv(nama_berkas, index=False)
    cek = pd.read_csv(nama_berkas)
    print(f"[simpan]    {nama_berkas}: {len(cek)} baris tersimpan dan terbaca kembali")
    os.remove(nama_berkas)

simpan(tambah_fitur(bersihkan(muat())), "transaksi_bersih.csv")

Output yang Dihasilkan:

[muat]      8 baris mentah
[bersihkan] 7 baris setelah baris ganda dibuang
[fitur]     7 baris, 5 kolom
[simpan]    transaksi_bersih.csv: 7 baris tersimpan dan terbaca kembali

Mari kita bedah rangkaian tersebut:

1. simpan(tambah_fitur(bersihkan(muat())), ...) adalah pipeline itu sendiri: keluaran muat() menjadi masukan bersihkan(), keluarannya menjadi masukan tambah_fitur(), dan seterusnya — urutan pembacaan dari dalam ke luar persis mengikuti urutan kerja. Menambah tahap baru berarti menulis satu fungsi dan menyisipkannya di satu tempat, bukan menyunting lima salinan.
2. Setiap fungsi hanya melihat masukannya. bersihkan() tidak tahu data berasal dari mana; tambah_fitur() tidak tahu baris ganda pernah dibuang. Keterpisahan inilah yang membuat setiap tahap bisa diuji sendirian dengan masukan kecil buatan.
3. .copy() di awal bersihkan() dan tambah_fitur() memastikan DataFrame masukan tidak berubah di tempatnya — fungsi tetap murni meskipun di dalamnya kita menambah kolom.
4. Efek samping dipenjara di tahap akhir: hanya simpan() yang menyentuh berkas. Ia juga memverifikasi roundtrip — menulis lalu membaca kembali — dan menghapus berkas contoh supaya folder proyek tetap bersih; pada proyek sungguhan, baris os.remove() tentu tidak ada.
5. Baris [fitur] ... 5 kolom menegaskan bahwa dua kolom baru (hari, total_ribu) lahir di tahap tiga, bukan menempel diam-diam di tengah proses — ringkasan per tahap membuat pipeline bisa diaudit dari keluarannya saja.

Pipeline scikit-learn: Saat Model Ikut Terlibat #

Pipeline fungsi murni menuntasi tahap data; begitu model masuk, kita memakai Pipeline milik scikit-learn. Alasannya bukan keindahan, melainkan kebocoran: imputer dan scaler harus di-fit hanya pada data latih, lalu diterapkan ke data uji. Bila kita menstandarkan seluruh data sebelum membaginya, rata-rata dan simpangan data uji ikut membocorkan diri ke data latih — kesalahan klasik yang sudah kita bahas pada chapter Mencegah Data Leakage . Pipeline memastikan seluruh langkah di-fit ulang setiap kali hanya pada bagian data latih, termasuk saat dijalankan di dalam validasi silang.

import numpy as np
import pandas as pd
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler

rng = np.random.default_rng(42)
n = 24
kunjungan = rng.integers(2, 12, n).astype(float)
umur_bulan = rng.integers(1, 25, n).astype(float)
berhenti = (kunjungan + rng.uniform(-1.5, 1.5, n) < 5).astype(int)
kunjungan[[3, 17]] = np.nan  # dua nilai hilang: tugas imputer

fitur = pd.DataFrame({"kunjungan": kunjungan, "umur_bulan": umur_bulan})

X_latih, X_uji, y_latih, y_uji = train_test_split(
    fitur, berhenti, test_size=6, stratify=berhenti, random_state=42,
)

pipeline = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),
    ("skala", StandardScaler()),
    ("model", LogisticRegression()),
])
pipeline.fit(X_latih, y_latih)

print("langkah pipeline :", [nama for nama, _ in pipeline.steps])
print("nilai terisi      : median kunjungan latih =",
      pipeline.named_steps["imputer"].statistics_[0])
print("akurasi data uji  :", f"{pipeline.score(X_uji, y_uji):.2f}")

Output yang Dihasilkan:

langkah pipeline : ['imputer', 'skala', 'model']
nilai terisi      : median kunjungan latih = 8.0
akurasi data uji  : 0.83

Tiga hal patut dicermati dari hasil tersebut:

1. Urutan langkah terjaga eksplisit — pipeline.steps menampilkannya sebagai daftar bernama: imputer, skala, model. Nama langkah bisa dipanggil kembali lewat named_steps, seperti saat kita membaca median yang dipakai imputer untuk menambal dua nilai hilang.
2. Median dihitung dari data latih saja (statistics_[0] = 8.0) — inilah jaminan anti-kebocoran yang menjadi alasan utama memakai Pipeline: bila kode ini dijalankan dalam validasi silang, imputer di-fit ulang pada tiap lipatan data latih.
3. fit() satu kali menjalankan semuanya — imputer di-fit, scaler di-fit pada hasil imputer, model di-fit pada hasil scaler; score() menerapkan ketiganya berurutan pada data uji tanpa menyentuh proses fitting.

Kapan Memakai yang Mana #

Keduanya bukan pesaing, melainkan tetangga: pipeline fungsi murni mengantarkan data mentah sampai ke pintu model, dan Pipeline scikit-learn mengambil alih sejak titik itu.

Rangkuman #

• Pipeline merangkai pembersihan dan transformasi menjadi rangkaian yang bisa diulang; menambah tahap berarti menambah satu fungsi di satu tempat, bukan menyalin kode ke banyak berkas.
• Fungsi murni — keluaran hanya dari masukan, tanpa efek samping — adalah blok bangunannya; efek samping seperti menulis berkas dipisahkan ke tahap paling akhir.
• Pipeline contoh kita memuat, membersihkan, menambah fitur, lalu menyimpan — setiap tahap mencetak ringkasan sehingga bentangan data terpantau dari luar.
• Saat model terlibat, Pipeline scikit-learn menjaga imputer dan scaler di-fit hanya pada data latih, menutup celah data leakage terutama dalam validasi silang.

Pipeline kita kini menghasilkan keluaran yang sama dari masukan yang sama — setidaknya pada mesin yang sama, minggu yang sama. Pertanyaan berikutnya menyeberang lebih jauh: bisakah orang lain, di komputer lain, bulan depan, memperoleh angka yang persis sama? Menjaga agar jawabannya “ya” adalah pekerjaan chapter Reproducible Analysis .

Referensi #

1. scikit-learn — Cross-validation: evaluating estimator performance