Pengujian dalam pipeline

Pipeline yang mengolah data dari sumber yang tidak sepenuhnya terkendali memerlukan pemeriksaan yang dijalankan mesin pada setiap run, bukan pemeriksaan yang bergantung pada keberuntungan. Topik ini membahas tiga pola pengujian di dalam pipeline, yaitu aserasi sebagai gerbang yang menghentikan proses, ekspektasi yang dinyatakan terpisah dari kode transformasi, dan pola karantina yang memindahkan baris gagal tanpa menghentikan baris lain. Satu demonstrasi dengan berkas CSV kecil dijalankan sungguhan, sehingga pembagian antara baris lolos dan baris terkarantina terlihat pada keluaran aslinya. Setelah membacanya, pilihan antara menghentikan pipeline atau melanjutkan pemrosesan dapat diputuskan per aturan, bukan secara subjektif.

Aserasi sebagai gerbang pipeline#

Aserasi (assertion) adalah pemeriksaan yang menyatakan sebuah keharusan, yaitu kondisi yang wajib terpenuhi agar pemrosesan boleh berlanjut. Pada Python, pemeriksaan ini ditulis sebagai pernyataan assert yang diikuti pesan penjelasnya, dan program yang menemukan kondisi terlanggar langsung berhenti dengan error berjenis AssertionError. Perilaku berhenti itulah yang disebut gerbang, karena baris yang tidak layak tidak pernah sampai ke tahap berikutnya dan penyebabnya tercatat pada pesan aserasinya.

# tidak-dijalankan
assert baris["id_pesanan"].strip() != "", "id_pesanan wajib terisi"
assert 1 <= int(baris["jumlah_item"]) <= 100, "jumlah_item di luar rentang 1-100"

Posisi gerbang menentukan manfaatnya. Aserasi yang ditempatkan tepat setelah data masuk menghentikan cacat sebelum nilai itu dipakai transformasi mana pun, sehingga biaya perbaikannya paling murah. Sebaliknya, gerbang yang menuntut kesempurnaan penuh pada data yang wajar saja cacat membuat pipeline gagal terus-menerus, dan kegagalan itu lambat laun dianggap hal biasa. Karena itu, aserasi dipilih untuk pelanggaran yang membuat baris tidak mungkin diproses dengan benar, misalnya kunci utama yang kosong, sedangkan aturan yang masih dapat ditoleransi diserahkan ke pola karantina.

Ekspektasi terpisah dari transformasi#

Ekspektasi (expectation) adalah aturan kualitas yang dinyatakan sebagai konfigurasi terpisah, bukan sebagai percabangan yang diselipkan di dalam kode transformasi. Bentuk deklaratif ini membuat aturan dapat diberi nama, dibaca ulang, dan dikelola versinya tanpa menyentuh logika transformasinya, sebagaimana kode transformasi dikelola secara terpisah dari data yang diprosesnya. Satu aturan yang sama juga dapat diterapkan pada tabel lain, sehingga pengujian kualitas tumbuh menjadi kumpulan aturan yang terstruktur, bukan kumpulan baris kode yang menyebar di banyak skrip.

Great Expectations adalah library Python yang mewujudkan pola tersebut. Aturan ditulis sebagai ekspektasi bernama, misalnya nilai kolom tidak boleh kosong atau nilai kolom harus berada pada rentang tertentu, kemudian pustaka menilai data terhadap kumpulan ekspektasinya dan menyusun laporan hasilnya. Cuplikan berikut memperlihatkan bentuk deklaratifnya, dan pustakanya dipasang terpisah karena tidak tersedia di lingkungan kerja jalur ini.

# tidak-dijalankan
"""Ekspektasi dinyatakan sebagai objek deklaratif, terpisah dari transformasi.

Pustaka Great Expectations dipasang terpisah: pip install great-expectations
"""
import great_expectations as gx

context = gx.get_context()

suite = context.suites.add(gx.ExpectationSuite(name="pesanan_bersih"))
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToNotBeNull(column="id_pesanan")
)
suite.add_expectation(
    gx.expectations.ExpectColumnValuesToBeBetween(
        column="jumlah_item", min_value=1, max_value=100
    )
)

Pola kedua ini berbeda dari aserasi pada empat aspek yang sering menjadi pertimbangan.

Perbandingan itu bukan pilihan yang saling menggantikan. Skrip pipeline yang kecil cukup dengan aserasi, sedangkan kumpulan tabel yang diuji banyak aturan sebaiknya menempatkan aturannya sebagai ekspektasi agar tetap terkendali.

Pola karantina#

Karantina (quarantine) adalah pola yang memindahkan baris yang gagal aturan ke tempat penyimpanan tersendiri, sehingga baris yang lolos tetap diproses dan pipeline tidak berhenti. Tempat karantinanya dapat berupa tabel, berkas CSV, atau antrean dead-letter, dan setiap barisnya disertai alasan penolakan serta penanda run agar dapat ditelusuri. Pola ini dipilih bila toleransi ada, misalnya satu baris cacat tidak perlu menahan seribu baris lain yang sehat, sedangkan gerbang aserasi tetap dipakai untuk kegagalan yang membuat seluruh run tidak bermakna. Bentuk paling sederhananya berupa satu baris CSV per pelanggaran, sedangkan pada skala yang lebih besar tabel karantina diberi partisi per tanggal agar pembersihannya terkendali.

Dua kebiasaan menjaga karantina agar tidak berubah menjadi pembuangan diam-diam. Pertama, jumlah baris karantina per run diperlakukan sebagai angka yang dipantau, karena lonjakan tiba-tiba biasanya berarti ada perubahan di sumber. Kedua, isi karantina ditinjau berkala untuk memutuskan tindak lanjutnya. Tindak lanjut itu berupa memperbaiki data di sumbernya lalu memuat ulang, melonggarkan aturan yang ternyata terlalu ketat, atau menggagalkan run berikutnya bila persentasenya melewati ambang batas.

Demonstrasi aserasi dan karantina#

Demonstrasi berikut menjalankan kedua pola pada satu pekerjaan kecil. Berkas masuknya berisi tujuh baris pesanan dengan tiga baris cacat yang ditanamkan. Tiga cacat itu berupa satu id_pelanggan yang kosong, satu tanggal yang berumur di luar jendela kesegaran, dan satu baris dengan jumlah di luar rentang sekaligus pelanggan yang tidak terdaftar. Skrip validasi.py memeriksa setiap baris dengan empat fungsi aserasi, lalu memisahkan barisan lolos dan barisan terkarantina ke dua berkas CSV yang berbeda.

id_pesanan,tanggal_pesanan,id_pelanggan,jumlah_item
P201,2026-11-01,C01,3
P202,2026-10-15,C02,1
P203,2026-11-01,,2
P204,2026-09-10,C01,4
P205,2026-11-01,C99,0
P206,2026-10-28,C03,6
P207,2026-11-02,C04,10

"""Pola pengujian dalam pipeline: aserasi per aturan dan karantina baris gagal.

Setiap aturan berupa fungsi aserasi yang membaca satu baris dan melempar
AssertionError bila aturannya dilanggar. Baris yang lolos seluruh aturan
ditulis ke pesanan_bersih.csv dan pemrosesan dilanjutkan, sedangkan baris
yang gagal dipindahkan ke karantina.csv bersama alasan penolakannya.
"""

import csv
from datetime import date
from pathlib import Path

BERKAS_MASUK = "pesanan_masuk.csv"
BERKAS_BERSIH = "pesanan_bersih.csv"
BERKAS_KARANTINA = "karantina.csv"

KOLOM = ("id_pesanan", "tanggal_pesanan", "id_pelanggan", "jumlah_item")

HARI_INI = date(2026, 11, 2)     # tanggal literal agar keluaran deterministik
JENDELA_HARI = 30                # batas umur baris, dalam hari
RENTANG_JUMLAH = (1, 100)        # rentang jumlah_item yang valid
PELANGGAN_TERDAFTAR = {"C01", "C02", "C03", "C04"}


def asersi_id_wajib(baris: dict) -> None:
    """Aturan 1: id_pesanan wajib terisi."""
    assert baris["id_pesanan"].strip() != "", "id_pesanan kosong"


def asersi_jumlah_rasional(baris: dict) -> None:
    """Aturan 2: jumlah_item berupa bilangan bulat pada rentang yang valid."""
    assert baris["jumlah_item"].strip() != "", "jumlah_item kosong"
    jumlah = int(baris["jumlah_item"])
    assert (
        RENTANG_JUMLAH[0] <= jumlah <= RENTANG_JUMLAH[1]
    ), f"jumlah_item {jumlah} di luar rentang {RENTANG_JUMLAH[0]}-{RENTANG_JUMLAH[1]}"


def asersi_kesegaran(baris: dict) -> None:
    """Aturan 3: umur baris tidak melewati jendela kesegaran."""
    tanggal = date.fromisoformat(baris["tanggal_pesanan"])
    umur = (HARI_INI - tanggal).days
    assert (
        umur <= JENDELA_HARI
    ), f"tanggal_pesanan berumur {umur} hari di luar jendela {JENDELA_HARI} hari"


def asersi_pelanggan_terdaftar(baris: dict) -> None:
    """Aturan 4: id_pelanggan wajib terisi dan terdaftar pada tabel acuan."""
    id_pelanggan = baris["id_pelanggan"].strip()
    assert id_pelanggan != "", "id_pelanggan kosong"
    assert (
        id_pelanggan in PELANGGAN_TERDAFTAR
    ), f"id_pelanggan {id_pelanggan} tidak terdaftar"


ATURAN = (asersi_id_wajib, asersi_jumlah_rasional, asersi_kesegaran,
          asersi_pelanggan_terdaftar)


def periksa_baris(baris: dict) -> list[str]:
    """Jalankan seluruh aturan pada satu baris dan kumpulkan alasannya.

    Aserasi menjadi gerbang: setiap pelanggaran ditangkap, alasan itu
    terkumpul, dan baris tidak pernah masuk tahap berikutnya begitu saja.
    """
    alasan = []
    for satu in ATURAN:
        try:
            satu(baris)
        except AssertionError as error:
            alasan.append(str(error))
    return alasan


def tulis_barisan(path: str, barisan: list[dict], dengan_alasan: bool) -> None:
    """Tulis barisan ke berkas CSV, ditambah kolom alasan bila dikarantina."""
    with open(path, "w", newline="", encoding="utf-8") as berkas:
        if dengan_alasan:
            penulis = csv.DictWriter(
                berkas, fieldnames=list(KOLOM) + ["alasan"]
            )
        else:
            penulis = csv.DictWriter(berkas, fieldnames=KOLOM)
        penulis.writeheader()
        penulis.writerows(barisan)


if __name__ == "__main__":
    with open(BERKAS_MASUK, newline="", encoding="utf-8") as berkas:
        barisan_masuk = list(csv.DictReader(berkas))

    barisan_lolos: list[dict] = []
    barisan_karantina: list[dict] = []
    for baris in barisan_masuk:
        alasan = periksa_baris(baris)
        if alasan:
            baris["alasan"] = "; ".join(alasan)
            barisan_karantina.append(baris)
        else:
            barisan_lolos.append(baris)

    tulis_barisan(BERKAS_BERSIH, barisan_lolos, dengan_alasan=False)
    tulis_barisan(BERKAS_KARANTINA, barisan_karantina, dengan_alasan=True)

    print(f"baris masuk : {BERKAS_MASUK} ({len(barisan_masuk)} baris)")
    print(f"lolos       : {len(barisan_lolos)} baris, dilanjutkan ke {BERKAS_BERSIH}")
    print(f"karantina   : {len(barisan_karantina)} baris, dipindahkan ke {BERKAS_KARANTINA}")
    print()
    print(f"Isi {BERKAS_BERSIH}:")
    print(Path(BERKAS_BERSIH).read_text(encoding="utf-8"), end="")
    print()
    print(f"Isi {BERKAS_KARANTINA}:")
    print(Path(BERKAS_KARANTINA).read_text(encoding="utf-8"), end="")

baris masuk : pesanan_masuk.csv (7 baris)
lolos       : 4 baris, dilanjutkan ke pesanan_bersih.csv
karantina   : 3 baris, dipindahkan ke karantina.csv

Isi pesanan_bersih.csv:
id_pesanan,tanggal_pesanan,id_pelanggan,jumlah_item
P201,2026-11-01,C01,3
P202,2026-10-15,C02,1
P206,2026-10-28,C03,6
P207,2026-11-02,C04,10

Isi karantina.csv:
id_pesanan,tanggal_pesanan,id_pelanggan,jumlah_item,alasan
P203,2026-11-01,,2,id_pelanggan kosong
P204,2026-09-10,C01,4,tanggal_pesanan berumur 53 hari di luar jendela 30 hari
P205,2026-11-01,C99,0,jumlah_item 0 di luar rentang 1-100; id_pelanggan C99 tidak terdaftar

Keluaran memperlihatkan pembagian yang tegas. Empat baris lolos seluruh aturan dan dilanjutkan ke pesanan_bersih.csv, sedangkan tiga baris cacat pindah ke karantina.csv tanpa menghentikan pemrosesan baris lainnya. Kolom alasan pada karantina.csv memuat pesan aserasi yang dilanggar, sehingga baris P205 yang melanggar dua aturan sekaligus menyimpan kedua alasannya dalam satu baris. Pada pekerjaan yang sesungguhnya, pesanan_bersih.csv diteruskan ke tahap transformasi, dan jumlah baris karantina dicatat pada pemantauan untuk ditinjau pada run berikutnya. Angka pada keluaran dapat dihitung ulang dari berkas masuknya, karena tiga baris cacat sengaja ditanamkan dan tanggal pembanding kesegaran ditetapkan sebagai literal.

Kesalahan umum#

1. Memilih gerbang aserasi untuk aturan yang datanya wajar saja cacat. Penyebabnya, seluruh aturan diperlakukan sama pentingnya sehingga semuanya diletakkan di depan. Akibatnya, pipeline gagal pada hampir setiap run dan kegagalan yang sungguh kritis tenggelam di antara kegagalan rutin. Perbaikannya, pisahkan aturan yang menghalangi pemrosesan dari aturan yang dapat ditoleransi, lalu salurkan kelompok kedua ke karantina.
2. Mengandalkan aserasi sebagai gerbang sementara program dijalankan dalam mode optimasi. Penyebabnya, Python membuang pernyataan assert ketika dijalankan dengan python3 -O, sehingga pemeriksaannya lenyap tanpa pesan apa pun. Akibatnya, baris cacat lolos ke tahap berikutnya seolah-olah seluruh aturan sudah diperiksa. Perbaikannya, tulis gerbang yang kritikal sebagai if yang melempar error secara eksplisit, atau pastikan pipeline tidak pernah dijalankan dengan opsi -O.
3. Menulis baris karantina tanpa alasannya. Penyebabnya, hanya isi baris asli yang disimpan agar skripnya lebih pendek. Akibatnya, tinjauan karantina tidak dapat memutuskan apa pun karena penyebab penolakan tidak tercatat. Perbaikannya, simpan pesan aserasi yang dilanggar pada kolom tersendiri, sebagaimana kolom alasan pada demonstrasi di atas.

Ringkasan#

• Aserasi adalah gerbang yang menghentikan pemrosesan bila keharusan dilanggar, dan dipakai untuk pelanggaran yang membuat baris tidak mungkin diproses dengan benar.
• Ekspektasi menyatakan aturan kualitas sebagai konfigurasi bernama yang terpisah dari kode transformasi, contoh mewujudkannya adalah Great Expectations.
• Pola karantina memindahkan baris gagal beserta alasannya sehingga pipeline tetap berjalan, dan karantina wajib dipantau jumlahnya serta ditinjau isinya secara berkala.

Referensi#

• The assert statement — Python documentation — Dokumentasi resmi Python tentang perilaku pernyataan assert, termasuk penonaktifannya pada mode optimasi.
• Documentation — Great Expectations — Dokumentasi Great Expectations, library yang menyatakan aturan kualitas sebagai ekspektasi terukur.