Otomasi Pemrosesan Data

Seluruh bahan sudah lengkap: tahap extract membaca banyak sumber, tahap transform merapikan data, gerbang validasi menolak data bermasalah, log mencatat perjalanan, dan ekspor menulis lima format tujuan. Yang tersisa adalah menyatukannya menjadi satu skrip yang dapat dijalankan dari terminal, dengan opsi baris perintah (command-line interface, CLI) sehingga penjadwal, rekan kerja, maupun Anda sendiri tidak perlu membuka kode untuk mengubah perilaku. Chapter ini membungkus pipeline Toko Berkah Abadi menjadi skrip CLI dengan module argparse, menambahkan pemrosesan per chunk untuk berkas masuk yang besar, dan menutup section ini dengan jembatan menuju penerapan nyata pada section Proyek.

Opsi Baris Perintah dengan argparse #

Module argparse mengubah argumen terminal menjadi objek yang enak dipakai. Skrip kami menyediakan dua opsi dengan nilai bawaan:

Nilai bawaan membuat skrip tetap berjalan ketika dipanggil tanpa argumen apa pun — syarat mutlak untuk penjadwal otomatis. Fungsi parse_argumen() juga menerima parameter argv berupa None, sehingga skrip membaca argumen terminal secara default dan tetap mudah diuji dengan daftar argumen buatan.

Skrip CLI Lengkap #

Perhatikan alur lengkapnya: berkas masuk ditulis ke TemporaryDirectory (simulasi berkas yang tiba dari sistem kasir), dibaca per chunk, ditransform per chunk, digabung, dibersihkan dari duplikat, lalu diringkas per cabang.

"""Skrip CLI pipeline ETL Toko Berkah Abadi, siap dijalankan terjadwal."""
import argparse
import tempfile
from pathlib import Path

import pandas as pd

TEKS_CSV = """id_pesanan,cabang,produk,jumlah,harga
P001,jakarta ,Kopi Arabika,2,45000
P002,Bandung,Teh Melati,3,18000
P003,Surabaya,Beras Premium,1,62000
P004,JAKARTA,Gula Pasir,4,14000
P005,Bandung,Kopi Arabika,2,45000
P005,Bandung,Kopi Arabika,2,45000
P006,Surabaya,Minyak Goreng,3,32000
"""


def parse_argumen(argv: list[str] | None = None) -> argparse.Namespace:
    """Baca opsi CLI dengan nilai bawaan agar skrip jalan tanpa argumen."""
    parser = argparse.ArgumentParser(description="Pipeline ETL Toko Berkah Abadi")
    parser.add_argument("--chunksize", type=int, default=4,
                        help="jumlah baris per chunk saat membaca CSV")
    parser.add_argument("--cabang", default=None,
                        help="proses satu cabang saja, misalnya Jakarta")
    return parser.parse_args(argv)


def transform_chunk(chunk: pd.DataFrame) -> pd.DataFrame:
    """Transformasi satu chunk: rapikan cabang dan hitung total."""
    df = chunk.assign(cabang=chunk["cabang"].str.strip().str.title())
    return df.assign(total=df["jumlah"] * df["harga"])


def muat_chunks(path: Path, chunksize: int) -> pd.DataFrame:
    """Tahap extract untuk berkas besar: proses per chunk lalu gabungkan."""
    bagian: list[pd.DataFrame] = []
    for nomor, chunk in enumerate(pd.read_csv(path, chunksize=chunksize), start=1):
        bagian.append(transform_chunk(chunk))
        print(f"chunk {nomor}      : {len(chunk)} baris mentah")
    return pd.concat(bagian, ignore_index=True)


if __name__ == "__main__":
    args = parse_argumen()
    print(f"argumen  : chunksize={args.chunksize}, "
          f"cabang={args.cabang or 'semua'}")
    with tempfile.TemporaryDirectory() as td:
        sumber = Path(td) / "pesanan_masuk.csv"
        sumber.write_text(TEKS_CSV, encoding="utf-8")
        df = muat_chunks(sumber, args.chunksize)

    df = df.drop_duplicates(subset="id_pesanan")
    if args.cabang is not None:
        df = df.loc[df["cabang"] == args.cabang]

    ringkasan = (
        df.groupby("cabang", as_index=False)
        .agg(transaksi=("id_pesanan", "count"), omzet=("total", "sum"))
        .sort_values("omzet", ascending=False)
        .reset_index(drop=True)
    )
    print(f"gabungan : {len(df)} baris unik")
    print(ringkasan.to_string(index=False))
    print("Pipeline selesai.")

Output yang Dihasilkan:

argumen  : chunksize=4, cabang=semua
chunk 1      : 4 baris mentah
chunk 2      : 3 baris mentah
gabungan : 6 baris unik
  cabang  transaksi  omzet
Surabaya          2 158000
 Jakarta          2 146000
 Bandung          2 144000
Pipeline selesai.

Bagian yang layak dibedah:

1. parse_argumen() membangun ArgumentParser dan dua opsi. Ketika skrip dijalankan tanpa argumen, parse_args() mengembalikan Namespace berisi nilai bawaan dan program berjalan normal dengan kode keluar 0.
2. muat_chunks() memanggil pd.read_csv(path, chunksize=...). Hasilnya bukan DataFrame melainkan iterator; loop memproses setiap chunk dengan transform_chunk() dan mengumpulkannya di list.
3. transform_chunk() menjalankan pembersihan pada potongan kecil — str.strip().str.title() untuk cabang dan perkalian total — sehingga memori tetap rendah karena seluruh berkas tidak pernah dimuat sekaligus.
4. Tahap akhir menggabungkan chunk dengan pd.concat(), membuang duplikat P005 dengan drop_duplicates(subset="id_pesanan"), menerapkan filter --cabang bila diberikan, dan merangkum omzet dengan groupby().

Output default menunjukkan dua chunk (4 dan 3 baris) yang setelah penggabungan dan deduplikasi menghasilkan enam baris unik, dengan Surabaya memimpin omzet 158000. Menjalankan python3 cli_pipeline.py --chunksize 2 --cabang Jakarta menghasilkan empat chunk dan ringkasan yang hanya memuat baris Jakarta — perubahan perilaku tanpa satu baris kode pun disentuh.

Menjadwalkan Pipeline #

Skrip CLI seperti di atas dirancang untuk dijalankan sistem, bukan manusia. Tiga pilihan penjadwalan yang umum:

• cron (Linux/macOS): entri crontab seperti 0 6 * * * python3 /path/cli_pipeline.py menjalankan pipeline tiap pukul 06.00.
• Penjadwal sistem lain: Task Scheduler di Windows atau systemd timer di Linux menawarkan kemampuan setara dengan integrasi log sistem.
• Orchestrator (misalnya Apache Airflow): dipakai ketika pipeline menjadi graf dependensi yang panjang dan butuh pemantauan, percobaan ulang, serta backfill; untuk pipeline satu berkas seperti ini, cron sudah memadai.

Apa pun pilihannya, syarat skripnya sama: kode keluar 0 saat sukses dan bukan nol saat gagal, log yang informatif, serta perilaku yang idempoten — menjalankan ulang pada hari yang sama tidak menggandakan data, dan itulah alasan replace dipilih pada chapter ekspor.

Section Praktik Data Engineering berakhir di sini dengan satu kesimpulan: pipeline pandas yang baik bukan sekadar rangkaian method, melainkan fungsi-fungsi kecil yang diuji, divalidasi, dicatat, dan dibungkus untuk dijalankan ulang. Saatnya menerapkan seluruh pola tersebut pada data yang utuh dan keputusan analisis yang nyata — masuklah ke section Proyek dan mulai dari Proyek Analisis Penjualan, Bagian 1: Memahami Dataset Penjualan.

Referensi resmi: pandas.read_csv , pandas.DataFrame.groupby , dan user guide IO .