Skrip sekali jalan memang menyelesaikan pekerjaan hari ini — tetapi besok data mentah baru datang, dan minggu depan lagi. Pipeline adalah jawaban untuk pekerjaan yang berulang: rangkaian tahap bernama yang membaca sumber, memvalidasi, membersihkan, mentransformasi, lalu menuliskan hasil dengan cara yang sama setiap dijalankan. Seluruh bab section ini — membaca CSV, validasi JSON, pembersihan, transformasi, join — adalah komponennya; chapter ini merakitnya menjadi satu program utuh yang bisa dijalankan dari CLI, diamati lewat log, dan diuji dengan pytest.

Anatomi Enam Tahap #

Pipeline kita mengikuti pola ETL yang diperluas:

Setiap tahap adalah fungsi yang menerima data dan mengembalikan data — bentuk masukan dan keluaran yang konsisten inilah yang membuat tahap bisa disusun seperti pipa (pola komposisi dari chapter Function Composition). Bahan mentahnya adalah penjualan_mentah.csv berikut, yang menumpuk masalah dari semua chapter sebelumnya:

text 
tanggal,sku,qty,harga
2025-06-13,P001,2,85000
2025-06-13,p005,3,15000
2025-06-13,P099,1,12000
2025-06-14,P002,1,45000
2025-06-14,P002,1,45000
2025-06-15,P001,2,85000
2025-06-15,P008,10,3500
2025-06-15,P003,,32000
2025-06-16,P005,5,Rp 15.000
2025-06-16,P004,1,78000
2025-06-17,P005,2,15000
2025-06-17,P010,1,250000

Ada p005 dengan kapital salah, P099 yang tidak ada di master produk, P002 terkirim dua kali, qty kosong, dan harga berformat Rp 15.000. Pipeline juga membaca produk.csv dan kategori.csv dari chapter-chapter sebelumnya untuk validasi rujukan dan pengayaan data.

Kode Pipeline Utuh #

Simpan sebagai pipeline.py. Setiap tahap melaporkan dirinya lewat module logging — pesan progres inilah yang membuat pipeline bisa diawasi tanpa membuka debugger:

python 
"""Pipeline penjualan: ekstrak -> validasi -> bersih -> transformasi -> agregasi -> muat."""

import argparse
import csv
import json
import logging
import sys
from collections import Counter
from pathlib import Path

logging.basicConfig(stream=sys.stdout, level=logging.INFO,
                    format="%(levelname)-7s %(message)s")
log = logging.getLogger("pipeline")

KOLOM_WAJIB = ("tanggal", "sku", "qty", "harga")


def ekstrak(path_input):
    with open(path_input, newline="", encoding="utf-8") as berkas:
        mentah = list(csv.DictReader(berkas))
    log.info("Ekstrak    : %d baris dari %s", len(mentah), path_input)
    return mentah


def muat_sku_valid(path_produk):
    with open(path_produk, newline="", encoding="utf-8") as berkas:
        return {b["id"] for b in csv.DictReader(berkas)}


def validasi(mentah, sku_valid):
    valid, ditolak = [], []
    for nomor, baris in enumerate(mentah, start=2):
        if not all(baris.get(kolom, "").strip() for kolom in KOLOM_WAJIB):
            ditolak.append((nomor, "kolom kosong"))
        elif baris["sku"].upper() not in sku_valid:
            ditolak.append((nomor, f"sku tidak dikenal: {baris['sku']}"))
        elif not baris["qty"].strip().isdigit():
            ditolak.append((nomor, f"qty bukan angka: {baris['qty']!r}"))
        else:
            valid.append(baris)
    for nomor, alasan in ditolak:
        log.warning("Baris %d ditolak: %s", nomor, alasan)
    log.info("Validasi   : %d valid, %d ditolak", len(valid), len(ditolak))
    return valid


def bersihkan(valid):
    hasil = []
    for baris in valid:
        hasil.append({
            "tanggal": baris["tanggal"],
            "sku": baris["sku"].upper(),
            "qty": int(baris["qty"]),
            "harga": int(baris["harga"].replace("Rp", "").replace(".", "").strip()),
        })
    sebelum = len(hasil)
    unik = {(b["tanggal"], b["sku"], b["qty"], b["harga"]): b for b in hasil}
    log.info("Bersih     : %d baris, %d duplikat dihapus", len(unik), sebelum - len(unik))
    return list(unik.values())


def transformasi(bersih, path_produk, path_kategori):
    with open(path_produk, newline="", encoding="utf-8") as berkas:
        indeks_produk = {b["id"]: b for b in csv.DictReader(berkas)}
    with open(path_kategori, newline="", encoding="utf-8") as berkas:
        diskon = {b["kategori"]: int(b["diskon_persen"]) for b in csv.DictReader(berkas)}

    kaya = []
    for b in bersih:
        produk = indeks_produk[b["sku"]]
        kategori = produk["kategori"]
        total = b["qty"] * b["harga"]
        potongan = diskon.get(kategori, 0)
        kaya.append({
            **b,
            "nama": produk["nama"],
            "kategori": kategori,
            "total": total,
            "total_akhir": round(total * (1 - potongan / 100)),
        })
    log.info("Transform  : %d baris diperkaya nama, kategori, dan diskon", len(kaya))
    return kaya


def agregasi(kaya):
    per_kategori = {}
    for b in kaya:
        stat = per_kategori.setdefault(b["kategori"], {"jumlah": 0, "qty": 0, "omzet": 0})
        stat["jumlah"] += 1
        stat["qty"] += b["qty"]
        stat["omzet"] += b["total_akhir"]

    omzet_sku = Counter()
    for b in kaya:
        omzet_sku[b["sku"]] += b["total_akhir"]

    return {
        "total_transaksi": len(kaya),
        "omzet_keseluruhan": sum(b["total_akhir"] for b in kaya),
        "per_kategori": dict(sorted(per_kategori.items(), key=lambda x: -x[1]["omzet"])),
        "top_sku": [{"sku": sku, "omzet": total} for sku, total in omzet_sku.most_common(3)],
    }


def muat(kaya, ringkasan, dir_output):
    dir_output = Path(dir_output)
    dir_output.mkdir(parents=True, exist_ok=True)

    path_csv = dir_output / "penjualan_bersih.csv"
    with open(path_csv, "w", newline="", encoding="utf-8") as berkas:
        penulis = csv.DictWriter(berkas, fieldnames=list(kaya[0]))
        penulis.writeheader()
        penulis.writerows(kaya)
    log.info("Muat       : %d baris -> %s", len(kaya), path_csv)

    path_json = dir_output / "ringkasan.json"
    with open(path_json, "w", encoding="utf-8") as berkas:
        json.dump(ringkasan, berkas, ensure_ascii=False, indent=2)
        berkas.write("\n")
    log.info("Muat       : ringkasan -> %s", path_json)


def jalankan(path_input, path_produk, path_kategori, dir_output):
    mentah = ekstrak(path_input)
    valid = validasi(mentah, muat_sku_valid(path_produk))
    bersih = bersihkan(valid)
    kaya = transformasi(bersih, path_produk, path_kategori)
    ringkasan = agregasi(kaya)
    muat(kaya, ringkasan, dir_output)
    return ringkasan


def main():
    parser = argparse.ArgumentParser(description="Pipeline pembersihan data penjualan")
    parser.add_argument("--input", default="penjualan_mentah.csv")
    parser.add_argument("--produk", default="produk.csv")
    parser.add_argument("--kategori", default="kategori.csv")
    parser.add_argument("--output-dir", default="hasil")
    argumen = parser.parse_args()
    jalankan(argumen.input, argumen.produk, argumen.kategori, argumen.output_dir)


if __name__ == "__main__":
    main()

Mari kita bedah keputusan desain terpentingnya:

  1. Baris 63-67 (bersihkan): normalisasi upper(), koersi "Rp 15.000" menjadi 15000, lalu dedup dengan dictionary berkunci tuple seluruh kolom — pola dari chapter Membersihkan Data.
  2. Baris 20-22 (validasi): tiga aturan eksplisit — kolom lengkap, sku dikenal di master, qty numerik. Baris gagal ditolak dengan nomor barisnya, bukan didiamkan.
  3. Baris 96-101 (transformasi): join ke master produk (nama, kategori) dan master kategori (diskon), lalu menghitung total dan total_akhir — paduan pola join dari chapter Menggabungkan Dataset dan field turunan dari chapter Transformasi Data.
  4. Baris 134-135 (log pada muat): logging dengan lazy formatting %s menunda pemformatan string sampai benar-benar ditulis — praktik standar dari chapter Module logging.

Menjalankan Pipeline dari CLI #

Argumen command-line ditangani argparse (chapter Membuat CLI dengan argparse), sehingga pipeline bisa dijadwalkan otomatis:

bash 
python3 pipeline.py --input penjualan_mentah.csv --output-dir hasil

Output yang Dihasilkan:

text 
INFO    Ekstrak    : 12 baris dari penjualan_mentah.csv
WARNING Baris 4 ditolak: sku tidak dikenal: P099
WARNING Baris 9 ditolak: kolom kosong
INFO    Validasi   : 10 valid, 2 ditolak
INFO    Bersih     : 9 baris, 1 duplikat dihapus
INFO    Transform  : 9 baris diperkaya nama, kategori, dan diskon
INFO    Muat       : 9 baris -> hasil/penjualan_bersih.csv
INFO    Muat       : ringkasan -> hasil/ringkasan.json

Dua belas baris mentah menjadi sembilan baris bersih, dan setiap keputusan tercatat: baris 4 ditolak karena sku tak dikenal, baris 9 karena kolom kosong, satu duplikat dihapus. Hasil akhirnya dua berkas di direktori hasil/ — data bersih yang siap dipakai ulang:

text 
tanggal,sku,qty,harga,nama,kategori,total,total_akhir
2025-06-13,P001,2,85000,Kopi Gayo,minuman,170000,153000
2025-06-13,P005,3,15000,Keripik Singkong,camilan,45000,38250
2025-06-14,P002,1,45000,Teh Melati,minuman,45000,40500
2025-06-15,P001,2,85000,Kopi Gayo,minuman,170000,153000
2025-06-15,P008,10,3500,Mie Instan Goreng,bahan pokok,35000,33250
2025-06-16,P005,5,15000,Keripik Singkong,camilan,75000,63750
2025-06-16,P004,1,78000,"Beras Pandan Wangi, 5kg",bahan pokok,78000,74100
2025-06-17,P005,2,15000,Keripik Singkong,camilan,30000,25500
2025-06-17,P010,1,250000,Kopi Luwak,minuman,250000,225000
json 
{
  "total_transaksi": 9,
  "omzet_keseluruhan": 806350,
  "per_kategori": {
    "minuman": {
      "jumlah": 4,
      "qty": 6,
      "omzet": 571500
    },
    "camilan": {
      "jumlah": 3,
      "qty": 10,
      "omzet": 127500
    },
    "bahan pokok": {
      "jumlah": 2,
      "qty": 11,
      "omzet": 107350
    }
  },
  "top_sku": [
    {
      "sku": "P001",
      "omzet": 306000
    },
    {
      "sku": "P010",
      "omzet": 225000
    },
    {
      "sku": "P005",
      "omzet": 127500
    }
  ]
}

Idempotensi dan Titik Simpan #

Pipeline yang baik bersifat idempoten: menjalankannya dua kali menghasilkan keluaran yang sama, bukan data yang berlipat. Kita mendapatkannya secara gratis karena setiap tahap membaca ulang dari sumber dan menimpa keluaran lama dengan mode "w". Untuk menulis berkas keluaran secara aman dari gangguan (misalnya listrik padam di tengah penulisan), pola standarnya tulis dulu ke berkas sementara lalu ganti nama secara atomik:

python 
import os

def tulis_atomic(path, isi):
    path_tmp = path + ".tmp"
    with open(path_tmp, "w", encoding="utf-8") as berkas:
        berkas.write(isi)
    os.replace(path_tmp, path)

tulis_atomic("status.txt", "selesai")
print(open("status.txt", encoding="utf-8").read())

Output yang Dihasilkan:

text 
selesai

Fungsi os.replace() mengganti berkas lama dengan berkas baru dalam satu operasi sistem operasi — pembaca tidak pernah melihat berkas setengah jadi. Untuk pipeline multi-tahap yang mahal, konsep yang sama diperluas menjadi checkpoint: simpan keluaran antara (misalnya bersih.csv setelah tahap bersih) beserta jejak sumbernya, lalu lewati tahap yang masukannya belum berubah — biasanya dibandingkan lewat ukuran berkas atau hash.

Menguji Pipeline dengan pytest #

Pipeline yang dijadwalkan setiap malam butuh jaring pengaman: tes yang memastikan hasilnya tidak berubah diam-diam. Kita jalankan pipeline ke direktori uji, lalu bandingkan ringkasannya dengan nilai emas (golden values) yang sudah diverifikasi:

python 
import json

from pipeline import jalankan

def test_ringkasan_stabil():
    jalankan("penjualan_mentah.csv", "produk.csv", "kategori.csv", "hasil_test")

    with open("hasil_test/ringkasan.json", encoding="utf-8") as berkas:
        ringkasan = json.load(berkas)

    assert ringkasan["total_transaksi"] == 9
    assert ringkasan["omzet_keseluruhan"] == 806350
    assert ringkasan["top_sku"][0] == {"sku": "P001", "omzet": 306000}
    assert ringkasan["per_kategori"]["minuman"]["omzet"] == 571500

Simpan sebagai test_pipeline.py lalu jalankan pytest (chapter Menulis Test dengan pytest):

bash 
python3 -m pytest test_pipeline.py -q

Output yang Dihasilkan:

text 
.                                                                    [100%]
1 passed in 0.12s

Fungsi jalankan() dipanggil langsung tanpa subprocess — pipeline yang dirancang sebagai kumpulan fungsi murni per tahap otomatis mudah diuji. Jika sumber data berubah dan angka emas perlu diperbarui, tes gagal dengan pesan yang jelas; untuk pipeline besar, pola golden file membandingkan seluruh berkas keluaran byte-per-byte terhadap salinan yang telah diverifikasi manual.

Dari membaca satu file CSV sampai pipeline yang teruji — perjalanan section ini menutup satu lingkaran penuh. Regulasi berikutnya setelah ini adalah eksplorasi: saat fokus bergeser dari “membuat data layak pakai” ke “menemukan cerita di dalamnya”, section Data Science menanti dengan pandas, NumPy, dan visualisasi.