Membangun Pipeline Data

Setiap pagi, tim operasional Toko Berkah Abadi menutup laporan penjualan dari data pesanan yang masih mentah: penulisan nama cabang tidak seragam, kolom total belum ada, dan angkanya tersebar di berbagai berkas. Menyalin-tempel data secara manual setiap hari bukan cara kerja yang bisa dipertahankan, karena satu langkah yang terlewat langsung mengubah hasil laporan. Masalah seperti ini diselesaikan dengan pipeline data, yaitu rangkaian tahapan yang mengolah data mentah menjadi data siap analisis dan dapat dijalankan ulang setiap kali data baru tiba. Pada chapter ini kita menyusun pipeline ETL (extract, transform, load) dari fungsi-fungsi pandas kecil yang masing-masing punya satu tugas. Struktur yang dibangun di sini menjadi fondasi seluruh chapter pada section Praktik Data Engineering.

Anatomi Pipeline ETL #

Pipeline ETL membagi pekerjaan menjadi tiga tahap yang urutannya tetap. Data mengalir satu arah: sumber, proses perantara, lalu tujuan.

Aturan penyusunan yang kita pegang sederhana: satu fungsi menangani satu tahap, fungsi tersebut menerima DataFrame, dan mengembalikan DataFrame baru. Dengan kontrak seperti ini, tahapan bisa dipasang, dilepas, dan diuji secara terpisah tanpa saling menempel pada keadaan global. pandas 3.0 dengan Copy-on-Write yang selalu aktif mendukung gaya kerja ini, karena setiap operasi berperilaku seperti bekerja pada salinan data.

Implementasi Pipeline Pertama #

Script berikut menuntaskan ketiga tahap untuk data pesanan lima baris milik Toko Berkah Abadi. Perhatikan penulisan cabang yang masih berantakan pada tahap extract, lalu bandingkan dengan hasil tahap transform dan ringkasan akhir.

"""Pipeline ETL pertama: extract -> transform -> load untuk Toko Berkah Abadi."""
import io

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
"""


def extract(teks: str) -> pd.DataFrame:
    """Tahap extract: baca sumber menjadi DataFrame mentah."""
    return pd.read_csv(io.StringIO(teks))


def transform(df: pd.DataFrame) -> pd.DataFrame:
    """Tahap transform: rapikan cabang dan hitung kolom total."""
    df = df.assign(cabang=df["cabang"].str.strip().str.title())
    df = df.assign(total=df["jumlah"] * df["harga"])
    return df


def load(df: pd.DataFrame) -> pd.DataFrame:
    """Tahap load: susun ringkasan per cabang sebagai gudang data mini."""
    ringkasan = (
        df.groupby("cabang", as_index=False)
        .agg(transaksi=("id_pesanan", "count"), omzet=("total", "sum"))
        .sort_values("omzet", ascending=False)
        .reset_index(drop=True)
    )
    return ringkasan


if __name__ == "__main__":
    mentah = extract(TEKS_CSV)
    print("== Tahap extract ==")
    print(f"bentuk: {mentah.shape[0]} baris x {mentah.shape[1]} kolom")
    print(mentah.to_string(index=False))
    print()

    bersih = transform(mentah)
    print("== Tahap transform ==")
    print(bersih.to_string(index=False))
    print()

    hasil = load(bersih)
    print("== Tahap load ==")
    print(hasil.to_string(index=False))

Output yang Dihasilkan:

== Tahap extract ==
bentuk: 5 baris x 5 kolom
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

== Tahap transform ==
id_pesanan   cabang        produk  jumlah  harga  total
      P001  Jakarta  Kopi Arabika       2  45000  90000
      P002  Bandung    Teh Melati       3  18000  54000
      P003 Surabaya Beras Premium       1  62000  62000
      P004  Jakarta    Gula Pasir       4  14000  56000
      P005  Bandung  Kopi Arabika       2  45000  90000

== Tahap load ==
  cabang  transaksi  omzet
 Jakarta          2 146000
 Bandung          2 144000
Surabaya          1  62000

Mari kita telaah peran setiap fungsi:

1. extract(teks): fungsi read_csv() membaca sumber dan menghasilkan DataFrame. Argumen io.StringIO(TEKS_CSV) membuat string terlihat seperti berkas, sehingga contoh tetap bisa dijalankan tanpa menyimpan data ke disk.
2. transform(df): fungsi assign() menambahkan kolom baru dan mengembalikan DataFrame baru. Ekspresi df["cabang"].str.strip().str.title() membuang spasi sisa dan menyeragamkan kapitalisasi, sehingga JAKARTA dan jakarta sama-sama menjadi Jakarta.
3. load(df): fungsi groupby() mengelompokkan per cabang, lalu named aggregation pada agg() menghitung jumlah transaksi dan omzet. Pemanggilan sort_values() mengurutkan cabang dari omzet terbesar.
4. Blok if __name__ == "__main__": memanggil ketiga fungsi secara berurutan dan mencetak isi DataFrame pada setiap tahap, sehingga kita bisa memeriksa data berubah seperti apa setelah tiap tahap.

Nilai omzet Jakarta sebesar 146000 berasal dari pesanan P001 (90000) dan P004 (56000); angka ini muncul dengan benar meskipun penulisan cabangnya awalnya berbeda-beda. Bukti kecil inilah yang menunjukkan tahap transform bekerja.

Keuntungan Fungsi per Tahap #

Memecah pipeline menjadi fungsi kecil membawa tiga manfaat praktis:

• Mudah diuji: fungsi transform() bisa dipanggil dengan DataFrame kecil yang hasilnya sudah diketahui, tanpa perlu berkas asli.
• Mudah diganti: sumber CSV dapat diganti Parquet hanya dengan mengganti isi extract(), sementara tahap lain tidak tersentuh.
• Mudah dibaca ulang: setiap fungsi diberi docstring dan type hint pd.DataFrame, sehingga pembaca kode tahu input dan output yang diharapkan.

Pipeline di atas baru menerima satu sumber berupa string CSV. Data nyata biasanya tersebar di berkas CSV, Excel, JSON, dan database sekaligus; chapter berikutnya menyiapkan tahap extract yang mampu menarik data dari banyak sumber tersebut dan menggabungkannya menjadi satu DataFrame.

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