KAL
Menu

Pembangunan Pipeline

Pipeline ujung ke ujung: ingest, pembersihan, agregasi, dan penggabungan.

Daftar isi

Topik Pembangunan Pipeline mengerjakan inti proyek laporan lingkungan, yaitu mengubah log sensor yang kotor menjadi tabel ringkasan yang siap dianalisis. Seluruh pekerjaan dirangkai dalam satu skrip bernama pipeline.py, dan rangkaian semacam ini disebut pipeline, yakni deretan tahap pemrosesan yang dijalankan berurutan dengan keluaran satu tahap menjadi masukan tahap berikutnya. Empat tahapnya dijalankan dalam satu berkas yang sama, dan setiap tahap mencetak ringkasannya sendiri sehingga kesalahan berhenti pada tahap terjadinya. Pada akhir topik, berkas ringkasan_ruang.csv yang memuat satu baris per ruang sudah dihasilkan dan siap dianalisis pada topik Analisis dan Visualisasi.

Empat tahap dan alasan urutannya#

Empat tahap pipeline.py berjalan dalam urutan yang tidak boleh ditukar, yaitu ingest, pembersihan, agregasi, dan penggabungan. Tahap ingest membangun log kotor dan tabel meta, lalu menyimpannya sebagai berkas CSV seolah datanya baru datang dari sistem sumber. Tahap pembersihan membaca berkas itu kembali, merapikan nama ruang dengan aksesor .str, membuang baris duplikat dengan drop_duplicates, dan mengisi NaN suhu dengan median ruangnya melalui transform. Tahap agregasi meringkas baris yang sudah bersih menjadi satu baris per ruang dengan groupby, sedangkan tahap penggabungan menambahkan kolom meta, yaitu model sensor dan tanggal kalibrasi, melalui merge.

Urutan itu menentukan kebenaran hasil. Agregasi pada data kotor menghasilkan statistik yang menyesatkan karena baris duplikat memberi bobot ganda pada setiap perhitungan, sedangkan NaN membuat rata-rata dan simpangan baku bergeser. Nama ruang juga harus dirapikan lebih dahulu, karena "lab a" dan "Lab A" akan menjadi dua grup berbeda pada groupby sekaligus dua kunci yang berbeda pada merge, meski isinya sama. Pengisian NaN ditempatkan setelah duplikat dibuang agar median dihitung hanya dari baris yang sah.

Skrip pipeline.py#

Skrip berikut adalah pipeline.py utuh dengan empat blok tahap yang masing-masing diberi komentar berbahasa Indonesia. Fungsi bangun_log() memuat generator dataset beserta ketiga kekotorannya, sedangkan bangun_meta() memuat tabel pendamping satu baris per ruang untuk tahap penggabungan. Simpan berkas berikut pada folder kerja, lalu jalankan perintah pada blok bash di bawahnya.

python pipeline.py 
# pipeline.py: pipeline ujung ke ujung, yaitu ingest, bersih, agregasi, dan join.
import numpy as np
import pandas as pd

def bangun_log() -> pd.DataFrame:
    """Membangun log sensor 96 bacaan dengan kotoran deterministik."""
    rng = np.random.default_rng(42)
    n = 96
    df = pd.DataFrame({
        "waktu": pd.date_range("2026-01-05 07:00", periods=n, freq="15min"),
        "ruang": rng.choice(["Lab A", "Lab B", "Ruang Arsip"], n, p=[0.4, 0.35, 0.25]),
        "suhu_c": np.round(rng.normal(24, 1.5, n), 1),
        "lembap_pct": np.round(rng.normal(60, 5, n), 1),
        "status_sensor": rng.choice(["normal", "kalibrasi", "rusak"], n, p=[0.85, 0.10, 0.05]),
    })
    # kotoran 1: sebagian kecil sel suhu dan kelembapan menjadi NaN
    df.loc[rng.random(n) < 0.05, "suhu_c"] = np.nan
    df.loc[rng.random(n) < 0.05, "lembap_pct"] = np.nan
    # kotoran 2: nama ruang tidak konsisten pada empat baris
    df.loc[7, "ruang"] = "lab a"
    df.loc[23, "ruang"] = "LAB B"
    df.loc[41, "ruang"] = "ruang arsip"
    df.loc[67, "ruang"] = " Lab A "
    # kotoran 3: tiga baris digandakan
    return pd.concat([df, df.sample(3, random_state=42)], ignore_index=True)

def bangun_meta() -> pd.DataFrame:
    """Tabel pendamping satu baris per ruang untuk tahap join."""
    return pd.DataFrame({
        "ruang": ["Lab A", "Lab B", "Ruang Arsip", "Lobi"],
        "model": ["T-100", "T-100", "T-200", "T-300"],
        "tanggal_kalibrasi": pd.to_datetime(
            ["2025-12-01", "2025-12-10", "2025-11-20", "2025-10-05"]
        ),
    })

# Tahap 1: ingest, bangun data kotor lalu simpan sebagai CSV
log_kotor = bangun_log()
log_kotor.to_csv("log_sensor.csv", index=False)
meta = bangun_meta()
meta.to_csv("sensor_meta.csv", index=False)
print(f"ingest  : {len(log_kotor)} baris kotor ke log_sensor.csv, {len(meta)} baris meta ke sensor_meta.csv")

# Tahap 2: bersih, baca kembali lalu rapikan nama ruang, duplikat, dan NaN
log = pd.read_csv("log_sensor.csv", parse_dates=["waktu"])
print(f"bersih  : {len(log)} baris dibaca, {log['suhu_c'].isna().sum()} suhu dan {log['lembap_pct'].isna().sum()} kelembapan hilang")
log["ruang"] = log["ruang"].str.strip().str.title()
log = log.drop_duplicates()
log["median_ruang"] = log.groupby("ruang")["suhu_c"].transform("median")
log["suhu_c"] = log["suhu_c"].fillna(log["median_ruang"])
log = log.drop(columns="median_ruang")
print(f"bersih  : {len(log)} baris sesudah duplikat dibuang, {log['suhu_c'].isna().sum()} suhu tersisa hilang")

# Tahap 3: agregasi, ringkas statistik suhu per ruang
ringkasan = (
    log.groupby("ruang", as_index=False)
    .agg(
        suhu_rata=("suhu_c", "mean"),
        suhu_std=("suhu_c", "std"),
        jumlah_bacaan=("waktu", "size"),
    )
    .round(2)
)
print(ringkasan)

# Tahap 4: join, gabungkan ringkasan dengan meta yang kuncinya unik
laporan = ringkasan.merge(meta, on="ruang", how="left", validate="m:1")
laporan.to_csv("ringkasan_ruang.csv", index=False)
print(f"join    : {len(laporan)} baris laporan, {laporan['model'].isna().sum()} ruang tanpa meta")
print(laporan)
bash 
# menjalankan pipeline ujung ke ujung
python3 pipeline.py
text Keluaran 
ingest  : 99 baris kotor ke log_sensor.csv, 4 baris meta ke sensor_meta.csv
bersih  : 99 baris dibaca, 7 suhu dan 3 kelembapan hilang
bersih  : 96 baris sesudah duplikat dibuang, 0 suhu tersisa hilang
         ruang  suhu_rata  suhu_std  jumlah_bacaan
0        Lab A      23.72      1.28             37
1        Lab B      23.87      1.50             39
2  Ruang Arsip      24.30      1.72             20
join    : 3 baris laporan, 0 ruang tanpa meta
         ruang  suhu_rata  suhu_std  jumlah_bacaan  model tanggal_kalibrasi
0        Lab A      23.72      1.28             37  T-100        2025-12-01
1        Lab B      23.87      1.50             39  T-100        2025-12-10
2  Ruang Arsip      24.30      1.72             20  T-200        2025-11-20

Keluaran tersebut dibaca per tahap. Baris ingest menyatakan 99 baris kotor tersimpan bersama empat baris meta. Baris pembersihan pertama menunjukkan tujuh suhu dan tiga kelembapan hilang pada data yang dibaca, dan hanya suhu yang diisi karena laporan proyek ini tidak menggunakan kelembapan. Baris pembersihan kedua menunjukkan 96 baris tersisa dan nol suhu hilang, sesuai pembanding dari topik Penentuan Lingkup Proyek. Tabel agregasi memuat tiga baris dengan kolom suhu_rata, suhu_std, dan jumlah_bacaan, kemudian tahap penggabungan melengkapinya dengan model dan tanggal_kalibrasi hingga menjadi isi ringkasan_ruang.csv.

Satu baris meta tidak ikut muncul pada laporan, yaitu Lobi. Parameter how="left" mempertahankan kunci tabel kiri, dan Lobi tidak memiliki satu pun baris pada log, sehingga ruang itu wajar absen dari ringkasan. Perilaku ini sekaligus menunjukkan bahwa arah merge perlu dipilih sesuai tabel mana yang ingin dipertahankan.

Titik pemeriksaan antartahap#

Setiap tahap pada pipeline.py mencetak angka pembanding yang menjadi titik pemeriksaan ketika pipeline diubah atau diperluas. Jumlah baris turun dari 99 menjadi 96, yaitu turun tepat tiga sesuai jumlah duplikat yang disisipkan. Penurunan yang lebih besar berarti ada baris sah yang terbuang, sedangkan jumlah yang tidak turun berarti duplikat masih tersisa. Jumlah NaN suhu yang tetap di atas nol setelah tahap pembersihan berarti pengisian tidak berjalan, misalnya karena median dihitung sebelum nama ruang dirapikan sehingga pengisian tidak mengenai grupnya.

Pemeriksaan struktural berada pada parameter validate="m:1" di tahap penggabungan. Janji tersebut menyatakan kunci ruang pada tabel meta unik per ruang, dan pelanggarannya berhenti sebagai error MergeError, bukan sebagai tabel gabungan yang diam-diam melipatgandakan baris. Kebiasaan yang sama sudah dipraktikkan pada topik Kunci Join dan Validasi, dan di sini janji itu menjadi bagian tetap dari pipeline proyek.

Kesalahan umum#

  • Lupa parse_dates=["waktu"] pada read_csv. Kolom waktu terbaca sebagai teks, sehingga agregasi atau penyaringan berbasis tanggal tidak berjalan semestinya. Perbaikinya dengan menambahkan parameter tersebut saat membaca, seperti yang dilakukan skrip pada tahap pembersihan.
  • Menumpuk perubahan dengan inplace=True. Pada pandas 3.x parameter ini berstatus deprecated sehingga memicu peringatan dan praktiknya dihindari. Perbaikinya dengan menampung hasil operasi ke variabel, seperti pola yang digunakan pipeline.py.
  • Mengisi NaN suhu dengan satu angka untuk semua ruang, misalnya fillna(log["suhu_c"].median()). Median global mengaburkan perbedaan antar-ruang, sehingga karakter tiap ruang saling tercampur pada laporan. Perbaikinya dengan median per ruang melalui transform, sebagaimana dilakukan tahap pembersihan.

Ringkasan#

  • Pipeline empat tahap, yaitu ingest, pembersihan, agregasi, dan penggabungan, mengubah log kotor menjadi ringkasan per ruang dalam satu skrip yang dapat dijalankan ulang.
  • Urutan tahap menentukan kebenaran hasil: nama ruang dirapikan sebelum pengelompokan, duplikat dibuang sebelum statistik dihitung, dan NaN diisi setelahnya.
  • Angka pembanding per tahap, yaitu jumlah baris dan sisa NaN, serta validate="m:1", menjadikan kesalahan terlihat pada tahap asalnya.

Referensi#