Skrip pipeline yang ditulis sebagai satu blok panjang cepat menjadi sulit diuji: mengubah satu aturan pembersihan berisiko mengganggu bagian lain, dan logika yang sama tersalin ke banyak job. Fungsi memecah masalah itu menjadi unit kecil dengan masukan dan keluaran yang jelas, sedangkan modul mengelompokkan fungsi ke dalam berkas yang bisa diimpor oleh job mana pun. Data engineer juga rutin menulis aturan pembersihan yang dipakai berulang — misalnya mengubah teks harga "Rp 24.000" menjadi angka. Chapter ini membangun modul transformasi.py dari nol, menjalankannya langsung maupun lewat import, lalu menutup dengan packaging minimal.

Fungsi Murni untuk Transformasi Data #

Fungsi murni (pure function) menghasilkan keluaran yang hanya bergantung pada masukannya, tanpa efek samping. Fungsi bersih_harga("Rp 24.000") selalu mengembalikan 24000 berapa pun kali dipanggil. Fungsi yang menulis berkas atau mengubah isi list argumennya tidak murni — letakkan di tepi alur (baca dan muat), bukan di tengah logika transformasi.

Manfaatnya nyata di pipeline: fungsi murni mudah diuji, aman dijalankan ulang (job menjadi idempotent), dan aman dipanggil paralel. Aturan pembersihan yang sama tinggal sekali di satu modul, bukan tersalin di sepuluh job.

Modul transformasi.py #

Kita kumpulkan tiga fungsi bersih ke dalam satu modul. Berkas berikut utuh dan bisa langsung dieksekusi:

python 
"""Modul kecil untuk membersihkan data penjualan."""


def bersih_harga(nilai, default=0):
    """Ubah 'Rp 24.000' atau angka menjadi int rupiah."""
    if isinstance(nilai, (int, float)):
        return int(nilai)
    angka = "".join(karakter for karakter in str(nilai) if karakter.isdigit())
    return int(angka) if angka else default


def normalisasi_nama(nama):
    """Rapikan spasi berlebih dan kapitalisasi tiap kata."""
    return " ".join(str(nama).split()).title()


def ringkas(barisan):
    """Jumlahkan subtotal per produk dari list of dict."""
    hasil = {}
    for baris in barisan:
        produk = normalisasi_nama(baris.get("produk", "tanpa nama"))
        subtotal = bersih_harga(baris.get("harga", 0)) * baris.get("qty", 1)
        hasil[produk] = hasil.get(produk, 0) + subtotal
    return hasil


if __name__ == "__main__":
    # Blok ini hanya berjalan saat berkas dieksekusi langsung,
    # tidak berjalan saat transformasi di-import modul lain.
    print("Uji mandiri modul transformasi")
    print(bersih_harga("Rp 24.000"))
    print(bersih_harga("Rp15.000"))
    print(bersih_harga("N/A"))
    print(normalisasi_nama("  budi   santoso "))
    print(ringkas([
        {"produk": "kopi lampung", "qty": 2, "harga": "Rp 24.000"},
        {"produk": "teh melati", "qty": 3, "harga": "Rp15.000"},
    ]))

Jalankan berkas tersebut secara langsung:

bash 
python3 transformasi.py

Output yang Dihasilkan:

text 
Uji mandiri modul transformasi
24000
15000
0
Budi Santoso
{'Kopi Lampung': 48000, 'Teh Melati': 45000}

Mari kita telaah baris-per-baris kode di atas:

  1. Baris 4 (def bersih_harga(nilai, default=0)): parameter default memberi nilai cadangan bila masukan tidak mengandung angka, sehingga pemanggil tidak wajib mengisi argumen kedua.
  2. Baris 8 ("".join(...)): ekspresi generator menyaring karakter angka; "Rp 24.000" dan "Rp15.000" menjadi "24000" dan "15000".
  3. Baris 9: masukan tanpa digit seperti "N/A" menghasilkan default yaitu 0, bukan galat.
  4. Baris 21-22 (ringkas): modul boleh memakai fungsi miliknya sendiri; hasil.get(produk, 0) + subtotal mengakumulasi per produk.
  5. Baris 27 (if __name__ == "__main__":): saat dieksekusi langsung, __name__ bernilai "__main__" dan blok uji berjalan; saat diimpor, __name__ bernilai "transformasi" dan blok tersebut dilewati.

Import dari Skrip Lain #

Pemakaian sesungguhnya terjadi ketika job lain mengimpor modul tersebut. Skrip laporan.py berikut diletakkan satu folder dengan transformasi.py:

python 
"""Skrip utama: memakai modul transformasi lewat import."""

from transformasi import ringkas

penjualan = [
    {"produk": "kopi   lampung", "qty": 2, "harga": "Rp 24.000"},
    {"produk": "teh melati", "qty": 3, "harga": "Rp15.000"},
    {"produk": "gula aren", "qty": 1, "harga": 27000},
    {"produk": "gula aren", "qty": 2, "harga": 27000},
]

ringkasan = ringkas(penjualan)
print("Ringkasan penjualan")
for produk in sorted(ringkasan):
    print(f"  {produk:14} Rp{ringkasan[produk]}")
print("Total:", "Rp" + str(sum(ringkasan.values())))
bash 
python3 laporan.py

Output yang Dihasilkan:

text 
Ringkasan penjualan
  Gula Aren      Rp81000
  Kopi Lampung   Rp48000
  Teh Melati     Rp45000
Total: Rp174000

Mari kita telaah baris-per-baris kode di atas:

  1. Baris 3 (from transformasi import ringkas): hanya fungsi yang dibutuhkan yang diimpor; blok __main__ modul tidak berjalan — absennya baris “Uji mandiri” pada output membuktikannya.
  2. Baris 12: data dengan spasi ganda di "kopi lampung" tetap terkelompokkan berkat normalisasi_nama.
  3. Baris 15 ({produk:14}): format :14 meratakan kolom nama produk ke lebar 14 karakter tanpa library tambahan.

Parameter Default, *args, dan **kwargs #

Parameter default sudah terlihat pada bersih_harga. Dua mekanisme lainnya dipakai seperlunya: *args menampung argumen posisi berjumlah bebas, sedangkan **kwargs menampung argumen bernama. Di kode ETL, **kwargs paling sering dipakai untuk metadata opsional:

python 
def catatan(jam, sumber, **detail):
    tambahan = " ".join(f"{kunci}={nilai}" for kunci, nilai in sorted(detail.items()))
    return f"[{jam:02d}.00] {sumber} {tambahan}".rstrip()


print(catatan(9, "ingestion", baris=1200, status="selesai"))
print(catatan(10, "transform", durasi_detik=30))
print(catatan(11, "muat"))

Output yang Dihasilkan:

text 
[09.00] ingestion baris=1200 status=selesai
[10.00] transform durasi_detik=30
[11.00] muat

Fungsi catatan mengumpulkan argumen bernama ke dict detail; sorted(detail.items()) menjaga urutan keluaran tetap sama. *args jarang dibutuhkan pada kode transformasi — bila sebuah fungsi butuh banyak argumen posisi, biasanya fungsi tersebut perlu dipecah.

Packaging Minimal #

Ketika modul dipakai lintas proyek, letakkan di dalam package dengan tata letak src/ berikut:

text 
etl-penjualan/
├── pyproject.toml
└── src/
    └── etl/
        ├── __init__.py
        └── transformasi.py

Metadata build ditulis di pyproject.toml:

toml 
[project]
name = "etl-penjualan"
version = "0.1.0"
description = "Pipeline kecil untuk data penjualan"
requires-python = ">=3.12"

[build-system]
requires = ["setuptools"]
build-backend = "setuptools.build_meta"

Berkas __init__.py menandai folder sebagai package; setelah itu pip install -e . dari akar proyek memasang package dalam mode sunting, sehingga from etl.transformasi import ringkas bisa dipanggil dari job mana pun. Rujukan import dan package: https://docs.python.org/3/tutorial/modules.html .

Fungsi dan modul kini siap menjadi mesin transformasi. Mesin tersebut masih butuh bahan: data nyata yang dibaca dan ditulis sebagai berkas — dibahas di chapter berikutnya, Pemrosesan File.