Penyegaran Python untuk DE

Topik ini menyegarkan bagian Python yang paling sering dipakai pada pekerjaan data engineering, yaitu struktur skrip, penulisan fungsi, dan type hint. Pembahasan bergerak dari bentuk skrip yang aman diimpor, dilanjutkan dengan anotasi tipe pada fungsi, lalu ditutup dengan satu berkas transformasi utuh yang dijalankan bersama keluarannya. Setelah topik ini, kode pipeline dapat ditulis dengan bentuk yang mudah diuji, terbaca oleh alat statis, dan dipelihara dalam jangka panjang.

Struktur skrip dan gerbang main#

Skrip Python yang sehat untuk pipeline dapat dijalankan langsung dari terminal dan tetap aman diimpor sebagai modul. Kuncinya terletak pada variabel __name__ yang ditetapkan Python pada setiap modul. Apabila berkas dijalankan langsung dengan python3 transformasi.py, nilai __name__ adalah string __main__. Apabila berkas diimpor dengan import transformasi, nilai itu menjadi nama modulnya sendiri. Gerbang if __name__ == "__main__": memanfaatkan perbedaan tersebut untuk memisahkan definisi dari eksekusi.

Pemisahan ini bukan sekadar kerapian. Orkestrator (orchestrator) seperti Airflow, yaitu alat yang menjadwalkan dan menjalankan pipeline, mengimpor berkas pipeline sebagai modul, dan alat pengujian pun melakukan hal yang sama. Kode eksekusi di tingkat atas akan berjalan setiap kali impor terjadi, sehingga penjadwalan dan pengujian terganggu. Dengan gerbang __main__, impor hanya memuat fungsi dan definisi, sedangkan eksekusi nyata menunggu perintah eksplisit. Pola berikut memperlihatkan bentuk minimumnya.

def muat_catatan() -> list[str]:
    return ["buku,1", "pensil,2"]

def proses(catatan: list[str]) -> list[str]:
    return [c.upper() for c in catatan]

if __name__ == "__main__":
    for baris in proses(muat_catatan()):
        print(baris)

BUKU,1
PENSIL,2

Type hint untuk bentuk data yang jelas#

Type hint adalah anotasi tipe yang ditempelkan pada parameter fungsi, variabel, dan nilai kembalian. Pemeriksaan anotasi dilakukan oleh alat statis, bukan oleh Python saat program berjalan. Standarnya ditetapkan dalam PEP 484, sedangkan modul typing pada dokumentasi Python menjadi rujukan bentuk anotasinya. Untuk pekerjaan data engineering, type hint berperan sebagai kontrak antar tahap pipeline karena bentuk daftar, kamus, dan objek yang mengalir dari satu fungsi ke fungsi lain menjadi tertulis secara eksplisit. Anotasi juga membantu editor kode menampilkan saran yang tepat dan membantu pemeriksa statis seperti mypy menemukan ketidakcocokan tipe sebelum program dijalankan.

Contoh berikut menandai masukan sebagai daftar pasangan produk dan nilai, kemudian nilai kembalian sebagai kamus dengan kunci string dan nilai float.

def total_per_produk(catatan: list[tuple[str, float]]) -> dict[str, float]:
    hasil: dict[str, float] = {}
    for produk, nilai in catatan:
        hasil[produk] = hasil.get(produk, 0.0) + nilai
    return hasil

if __name__ == "__main__":
    sampel = [("buku", 4500.0), ("pensil", 2000.0), ("buku", 4500.0)]
    print(total_per_produk(sampel))

{'buku': 9000.0, 'pensil': 2000.0}

Bentuk generik seperti list[tuple[str, float]] pada contoh menunjukkan isi wadahnya secara bertingkat, sehingga pembaca tahu persis apa yang ada di dalam daftar itu. Ketika sebuah fungsi dapat gagal mengurai masukannya, nilai kembalian dianotasi dengan gabungan tipe, misalnya CatatanPenjualan | None, yang berarti hasil sah atau tidak ada hasil sama sekali. Dengan bentuk gabungan seperti ini, perilaku fungsi yang bisa gagal terbaca dari tanda tangannya, bukan tersembunyi di dalam tubuh fungsinya.

Berkas transformasi utuh#

Berkas berikut merangkum kebiasaan di atas menjadi satu modul transformasi kecil, lengkap dengan gerbang __main__ dan type hint di setiap fungsi. Modul ini membaca baris penjualan berformat CSV dari sebuah daftar, mengubah baris yang sah menjadi objek bertipe, membuang baris yang tidak valid, lalu mengagregasi total per produk. Dekorator @dataclass dari modul dataclasses dipakai untuk membuat kelas penampung data yang disebut dataclass. Dekorator itu menghasilkan __init__ dan representasi teks secara otomatis, sehingga definisi kelasnya tetap ringkas.

"""Transformasi ringkas catatan penjualan untuk kebutuhan pipeline."""
from dataclasses import dataclass


@dataclass
class CatatanPenjualan:
    """Satu baris penjualan yang sudah ternormalisasi."""

    produk: str
    jumlah: int
    harga_satuan: float

    @property
    def total(self) -> float:
        return self.jumlah * self.harga_satuan


def baca_baris(teks: str) -> CatatanPenjualan | None:
    """Mengubah satu baris CSV menjadi catatan, atau None bila tidak valid."""
    bagian = teks.strip().split(",")
    if len(bagian) != 3:
        return None
    produk, jumlah, harga = bagian
    try:
        return CatatanPenjualan(produk=produk, jumlah=int(jumlah), harga_satuan=float(harga))
    except ValueError:
        return None


def normalisasi(catatan: list[CatatanPenjualan]) -> list[CatatanPenjualan]:
    """Membuang catatan dengan jumlah nol atau negatif."""
    return [c for c in catatan if c.jumlah > 0]


def total_per_produk(catatan: list[CatatanPenjualan]) -> dict[str, float]:
    """Mengagregasi total penjualan per produk."""
    hasil: dict[str, float] = {}
    for c in catatan:
        hasil[c.produk] = hasil.get(c.produk, 0.0) + c.total
    return hasil


if __name__ == "__main__":
    baris_mentah = [
        "buku tulis,3,4500",
        "pensil,10,2000",
        "buku tulis,2,4500",
        "penghapus,0,1500",
        "spidol,d!!,3500",
    ]
    catatan = [c for c in (baca_baris(b) for b in baris_mentah) if c is not None]
    sah = normalisasi(catatan)
    for produk, total in total_per_produk(sah).items():
        print(f"{produk}: {total:.0f}")

buku tulis: 22500
pensil: 20000

Alur di dalam berkas mengikuti tahapan pipeline yang khas. Fungsi baca_baris mengembalikan None apabila baris tidak dapat diurai, misalnya baris spidol,d!!,3500 yang kolom jumlahnya bukan angka. Fungsi normalisasi kemudian membuang catatan dengan jumlah nol atau negatif, sehingga baris penghapus,0,1500 tidak ikut terhitung. Bagian terakhir mengagregasi sisa catatan per produk dan mencetak hasilnya, sedangkan nilai total dihitung oleh property total pada kelas CatatanPenjualan.

Bentuk seperti ini juga memudahkan penerimaan perubahan. Apabila aturan normalisasi berubah, misalnya jumlah nol tetap ikut dihitung, hanya fungsi normalisasi yang disentuh dan sisanya tidak terpengaruh. Pengujian otomatis dapat memanggil baca_baris dengan baris sampel tanpa menjalankan seluruh skrip karena gerbang __main__ menahan eksekusinya. Pada tim yang besar, type hint di setiap fungsi membuat peninjau kode dapat membaca kontrak data tanpa harus membaca seluruh isi fungsinya.

Kesalahan umum#

Struktur skrip dan type hint yang baru diterapkan sering kali tergelincir kembali ke pola lama, terutama saat kode lama dipindahkan ke proyek baru. Tiga kesalahan berikut paling sering muncul pada kode pipeline beserta perbaikannya.

• Eksekusi ditaruh di tingkat atas modul, sehingga setiap import ikut menjalankan pipeline dan alat pengujian ikut menanggung bebannya. Perbaikannya, pindahkan seluruh eksekusi ke dalam gerbang __main__.
• Type hint hanya dipasang pada sebagian fungsi, sehingga bentuk data pada fungsi lain tetap samar dan pemeriksa statis kehilangan jejaknya di tengah rantai pemanggilan. Perbaikannya, anotasi seluruh fungsi publik dan jalankan pemeriksaan statis secara berkala.
• Satu fungsi mengerjakan pembacaan, transformasi, dan penulisan sekaligus, sehingga pengujian dan penggunaan ulang menjadi sulit. Perbaikannya, pecah fungsi per tahap transformasi dan biarkan bagian __main__ yang merangkainya.

Ringkasan#

• Gerbang if __name__ == "__main__": memisahkan definisi dari eksekusi sehingga skrip aman diimpor maupun dijalankan langsung.
• Type hint mendokumentasikan bentuk data yang mengalir antar tahap pipeline dan memungkinkan pemeriksaan statis oleh alat seperti mypy.
• Anotasi tipe tidak dieksekusi Python, sehingga kepatuhan tipe tetap bergantung pada alat statis, bukan pada saat program berjalan.
• Fungsi kecil dengan satu tanggung jawab lebih mudah diuji dan dirangkai ulang daripada satu skrip panjang.

Referensi#

• typing — Dokumentasi resmi Python — Rujukan modul typing beserta bentuk anotasi yang didukung Python 3.
• PEP 484 — Type Hints — Spesifikasi type hint beserta latar desainnya.