Anatomi pipeline

Topik ini menjabarkan anatomi pipeline batch menjadi empat tahap yang berurutan, yaitu ekstrak, validasi, transformasi, dan muat. Setiap tahap dijelaskan tanggung jawabnya beserta kontrak data yang mengikatnya pada tahap sebelum dan sesudahnya. Pembahasan ditutup dengan pipeline kecil yang berjalan sungguhan, dari berkas CSV sumber sampai tabel analitik pada basis data DuckDB, lengkap dengan ringkasan keluaran tiap tahap.

Empat tahap dari sumber ke target#

Pipeline batch bergerak satu arah dari sistem sumber menuju tabel analitik, dan perjalanan itu dipecah menjadi empat tahap yang masing-masing memikul satu tanggung jawab. Tahap ekstrak (extract) menarik data dari sumber apa adanya, tahap validasi menyaring baris yang cacat sejak dini, tahap transformasi menyelaraskan bentuk dan tipe data, sedangkan tahap muat (load) menulis hasilnya ke target analitik. Pembagian ini membuat pipeline dapat diuji per bagian, karena setiap tahap menerima masukan yang jelas dan menghasilkan keluaran yang dapat diperiksa.

Kompasinya dapat dirangkum dalam satu tabel.

Diagram berikut merangkum alur empat tahap tersebut beserta ringkasan tanggung jawabnya.

Ukuran pembagian mengikuti ukuran pekerjaan. Pipeline kecil dapat memuat keempat tahap dalam satu skrip, sebagaimana contoh di bawah, sedangkan pipeline besar memecahnya menjadi pekerjaan terpisah yang disusun oleh orkestrator. Apa pun ukurannya, urutan tanggung jawabnya tetap sama, sehingga pola pikir yang dibangun pada skrip kecil tetap berlaku pada sistem yang jauh lebih besar.

Kontrak antartahap#

Kontrak antartahap adalah bentuk data yang disepakati antara satu tahap dan tahap berikutnya, dan pada pipeline kecil kontrak itu sesederhana daftar dictionary Python dengan kunci yang sama. Tahap ekstrak menghasilkan barisan dictionary dengan kunci id, tanggal_pinjam, dan jumlah_buku. Tahap validasi menjamin nilai jumlah_buku sudah bertipe bilangan bulat, sedangkan tahap transformasi mengganti kunci tanggal_pinjam menjadi tanggal. Selama kontrak itu dijaga, setiap tahap boleh ditulis, dibaca, dan diubah tanpa memeriksa isi tahap lain.

Validasi yang tegas membuat kontrak berikutnya lebih ringan. Setelah tahap validasi lolos, tahap transformasi dan muat berhak mengandaikan tipe datanya sudah benar, sehingga keduanya tidak perlu menulis percabangan pertahanan untuk nilai yang tak masuk akal. Baris yang ditolak tidak dibuang diam-diam, melainkan dilaporkan dengan alasannya, sebagaimana terlihat pada keluaran demo di bawah. Kebiasaan yang sama, pada skala besar, menjadi karantina data (quarantine) dan pencatatan baris gagal yang diuraikan pada topik penanganan error.

Aturan validasi yang lazim mencakup tipe yang benar, isi yang wajib ada, rentang nilai yang masuk akal, dan kunci yang tidak kembar. Untuk pipeline kecil, pengecekan tipe saja sudah menyelamatkan separuh masalah, sebagaimana baris P004 pada demo nanti yang jumlah bukunya tertulis sebagai kata. Aturan yang lebih ketat dapat ditambahkan seiring bertambahnya kebutuhan analitik.

Pipeline empat tahap dalam Python#

Keempat tahap diterjemahkan menjadi empat fungsi Python yang dipanggil berurutan, sehingga anatomi pada diagram terbaca langsung pada kode. Berkas sumbernya adalah catatan peminjaman buku perpustakaan kecil dengan lima baris, dan satu baris di antaranya sengaja cacat agar tahap validasi terlihat bekerja. Isi berkas sumbernya sebagai berikut.

id,tanggal_pinjam,jumlah_buku
P001,2026-10-01,2
P002,2026-10-01,1
P003,2026-10-02,3
P004,2026-10-02,dua
P005,2026-10-03,1

Skrip berikut membaca berkas itu, menolak baris P004 karena jumlah_buku-nya bukan angka, memecah kolom tanggal_pinjam menjadi kolom tanggal, lalu menulis sisanya ke tabel peminjaman pada berkas gudang.duckdb. Bagian __main__ di akhir skrip menjalankan keempat fungsi tahap secara berurutan.

"""Pipeline batch empat tahap: ekstrak, validasi, transformasi, muat."""

import csv

import duckdb

BERKAS_SUMBER = "peminjaman.csv"
BERKAS_TARGET = "gudang.duckdb"


def ekstrak(path: str) -> list[dict]:
    """Baca seluruh baris berkas sumber ke dalam memori."""
    with open(path, newline="", encoding="utf-8") as berkas:
        barisan = list(csv.DictReader(berkas))
    print(f"[ekstrak]      {len(barisan)} baris dibaca dari {path}")
    return barisan


def validasi(barisan: list[dict]) -> list[dict]:
    """Tolak baris yang tidak dapat ditafsir, misalnya angka yang rusak."""
    lolos: list[dict] = []
    for baris in barisan:
        try:
            baris["jumlah_buku"] = int(baris["jumlah_buku"])
        except ValueError:
            print(f"[validasi]     baris {baris['id']} ditolak: jumlah_buku bukan angka")
            continue
        lolos.append(baris)
    print(f"[validasi]     {len(lolos)} dari {len(barisan)} baris lolos")
    return lolos


def transformasi(barisan: list[dict]) -> list[dict]:
    """Ubah barisan baris ke bentuk yang diharapkan tabel target."""
    for baris in barisan:
        baris["tanggal"] = baris.pop("tanggal_pinjam")[:10]
    print(f"[transformasi] {len(barisan)} baris diselaraskan ke bentuk target")
    return barisan


def muat(barisan: list[dict]) -> None:
    """Tulis barisan baris ke tabel analitik pada basis data target."""
    con = duckdb.connect(BERKAS_TARGET)
    con.execute("DROP TABLE IF EXISTS peminjaman")
    con.execute(
        "CREATE TABLE peminjaman "
        "(id VARCHAR, tanggal VARCHAR, jumlah_buku INTEGER)"
    )
    con.executemany(
        "INSERT INTO peminjaman VALUES (?, ?, ?)",
        [(b["id"], b["tanggal"], b["jumlah_buku"]) for b in barisan],
    )
    jumlah = con.execute("SELECT count(*) FROM peminjaman").fetchone()[0]
    con.close()
    print(f"[muat]         {jumlah} baris tersimpan di {BERKAS_TARGET}")


if __name__ == "__main__":
    barisan = ekstrak(BERKAS_SUMBER)
    barisan = validasi(barisan)
    barisan = transformasi(barisan)
    muat(barisan)

[ekstrak]      5 baris dibaca dari peminjaman.csv
[validasi]     baris P004 ditolak: jumlah_buku bukan angka
[validasi]     4 dari 5 baris lolos
[transformasi] 4 baris diselaraskan ke bentuk target
[muat]         4 baris tersimpan di gudang.duckdb

Jejak keluarannya memperlihatkan penyusutan baris di setiap gerbang, yaitu lima baris dibaca, empat lolos, empat diselaraskan, dan empat tersimpan. Ringkasan satu baris per tahap adalah bentuk observabilitas (observability) termurah, yaitu kemampuan melihat kondisi sistem dari jejak yang ditinggalkannya, karena saat angka target menyimpang, jejak itu langsung menunjukkan tahap yang kehilangan baris. Berkas gudang.duckdb muncul di direktori kerja sebagai target analitiknya, dan tabel di dalamnya dapat diperiksa langsung dengan DuckDB.

Kesalahan umum#

1. Menyatukan validasi ke dalam transformasi. Penyebabnya, penggabungan terasa lebih hemat satu tahap. Akibatnya, baris cacat ikut tertransformasi dan error meledak jauh dari sumbernya, tanpa jejak baris mana yang buruk. Perbaikannya, pisahkan penolakan baris dari pengubahan bentuk, lalu catat setiap baris yang ditolak beserta alasannya.
2. Mengekstrak tanpa jendela data. Penyebabnya, tarikan diarahkan pada seluruh isi sumber setiap run. Akibatnya, pipeline makin lambat dan makin mahal seiring sumber bertambah, padahal analitik hanya membutuhkan data yang baru. Perbaikannya, batasi tarikan pada jendela waktu tertentu, misalnya satu hari, sebelum beralih ke pemuatan inkremental.
3. Memuat tanpa ringkasan baris. Penyebabnya, pernyataan print dianggap remeh untuk skrip kecil. Akibatnya, saat jumlah baris target menyimpang, tidak ada jejak untuk menemukan tahap yang bocor. Perbaikannya, cetak atau catat jumlah baris masuk dan keluar pada setiap tahap.

Ringkasan#

• Anatomi pipeline batch terdiri atas empat tahap berurutan, yaitu ekstrak, validasi, transformasi, dan muat, dengan satu tanggung jawab per tahap.
• Antar tahap terikat kontrak bentuk data, dan validasi yang tegas membuat tahap berikutnya berhak mengandaikan data sudah bersih.
• Ringkasan jumlah baris di setiap tahap adalah alat observabilitas termurah untuk menemukan tahap yang kehilangan data.

Referensi#

• Fundamentals of Data Engineering — O’Reilly — Buku Reis dan Housley yang memetakan siklus hidup data dari pembangkitan sampai penyajian.
• Data Engineering Zoomcamp — DataTalksClub — Kurikulum terbuka yang mempraktikkan pipeline batch dari ekstraksi sampai orkestrasi.
• INSERT statements — DuckDB — Dokumentasi resmi pernyataan INSERT DuckDB yang dipakai tahap muat pada demo.