KAL
Menu

Pembangunan

Membangun pipeline dari ingestion, model, orkestrasi, sampai pengujian.

Daftar isi

Pembangunan menerapkan rencana proyek peminjaman perpustakaan menjadi pipeline batch yang benar-benar berjalan di terminal. Topik ini merakit seluruh tahapan, yaitu ekstraksi dua sumber, validasi dengan aserasi, transformasi SQL dengan DuckDB, muat ke tabel fakta dan dimensi, dan penjadwalan dengan Airflow. Hasil akhirnya berupa satu berkas pipeline yang dapat dieksekusi ulang dengan keluaran yang sama persis, ditambah berkas DAG yang menjadwalkannya setiap hari.

Struktur proyek#

Struktur proyek memisahkan berkas kode, berkas data contoh, dan berkas penjadwalan supaya setiap perubahan punya tempat yang jelas. Pembagian ini mengikuti kebiasaan proyek akhir pada Data Engineering Zoomcamp yang menaruh kode, data, dan penjelasannya dalam satu folder proyek. Batas perannya tegas, yaitu kode membaca data dan tidak pernah sebaliknya. Susunan folder proyek ini sebagai berikut.

text Struktur folder proyek 
pembangunan/
├── index.md              ← artikel topik ini
├── pipeline_proyek.py    ← pipeline batch utuh
├── peminjaman.csv        ← sumber 1, ekspor peminjaman
├── anggota.json          ← sumber 2, layanan anggota
└── dag_peminjaman.py     ← penjadwalan Airflow

Berkas pipeline_proyek.py menjadi satu-satunya berkas yang dijalankan, sedangkan dua berkas data bersifat masukan yang tidak pernah diubah oleh pipeline. Berkas DAG tidak dieksekusi pada lingkungan latihan karena membutuhkan instalasi Airflow, tetapi tetap disimpan agar proyeknya utuh. Pola berkas terpisah juga memudahkan pengujian karena data contoh dapat diganti tanpa menyentuh kode. Nama berkas dipilih deskriptif sehingga isi tiap berkas dapat ditebak dari namanya sebelum dibuka. Apabila proyek dipindahkan ke komputer lain, seluruh bagian berjalan kembali tanpa konfigurasi tambahan karena hanya bergantung pada Python dan DuckDB. Susunan inilah yang kelak menjadi tampilan awal portofolio saat folder proyek dibuka peninjau.

Ekstraksi dan validasi#

Ekstraksi membaca dua sumber ke memori sebagai daftar kamus, lalu validasi menghentikan pipeline sebelum data kotor sempat masuk ke transformasi. Fungsi ekstrak_csv membaca peminjaman.csv baris demi baris dan memetakan setiap baris ke kamus berdasarkan header, sedangkan fungsi ekstrak_json memuat anggota.json dengan modul json bawaan Python. Dua sumber dengan format berbeda tetap menyatu dalam satu bentuk data yang sama.

Validasi berupa aserasi (assertion), yaitu pemeriksaan yang memunculkan error dan menghentikan eksekusi apabila syaratnya tidak terpenuhi. Empat aserasi ditanam pada fungsi validasi, dua di antaranya memastikan kedua sumber tidak kosong dan setiap id_anggota pada CSV terdaftar pada JSON. Dua aserasi sisanya menolak tanggal kembali yang mendahului tanggal pinjam dan id_pinjam yang ganda. Aserasi ditempatkan sebelum transformasi karena cacat yang lolos akan menular ke seluruh tabel turunannya. Pemeriksaan semacam ini murah ditulis tetapi menyelamatkan waktu berhari-hari saat sumber berubah tanpa pemberitahuan.

Transformasi SQL dengan DuckDB#

Transformasi membangun tabel dimensi dan fakta di atas dua tabel staging menggunakan SQL yang dieksekusi DuckDB. Tabel staging diperlakukan apa adanya, dim_anggota menyalin anggota, dim_buku mengumpulkan pasangan judul dan kategori yang unik, dan fakta_peminjaman merangkai keduanya lengkap dengan durasi pinjam dalam hari. Fragmen SQL berikut memperlihatkan pembentukan tabel fakta.

Pemuatan staging memanfaatkan parameter terikat (parameterized query), yaitu pernyataan SQL dengan tanda tanya yang diisi nilai dari Python, sehingga penyusunan teks query secara manual terhindar. Cara ini menghindari error pemformatan sekaligus menjadi kebiasaan yang aman saat nilainya berasal dari luar proyek. SQL lalu mengambil alih karena logika relasi dan agregasi paling ringkas dinyatakan sebagai query, bukan sebagai perulangan Python.

sql Contoh: pembentukan tabel fakta 
-- tidak-dijalankan: pernyataan ini dijalankan lewat modul duckdb di pipeline_proyek.py
CREATE OR REPLACE TABLE fakta_peminjaman AS
SELECT
    p.id_pinjam,
    p.tanggal_pinjam,
    p.tanggal_kembali,
    date_diff('day', p.tanggal_pinjam, p.tanggal_kembali) AS durasi_hari,
    b.id_buku,
    a.id_anggota
FROM peminjaman_mentah p
JOIN dim_buku b USING (judul_buku, kategori)
JOIN dim_anggota a USING (id_anggota);

Kolom durasi_hari dihitung langsung di SQL dengan date_diff sehingga definisinya hidup di satu tempat. Id buku dihasilkan ROW_NUMBER atas urutan judul, dan pendekatan sederhana itu cukup karena daftar buku contoh tidak berubah. Pada proyek dengan buku yang terus bertambah, kunci buku perlu dipertahankan antar eksekusi, misalnya lewat tabel dimensi yang diperbarui bertahap.

Berkas pipeline lengkap#

Berkas berikut merangkai seluruh tahapan menjadi satu program yang dijalankan dengan perintah python3 pipeline_proyek.py. Urutan pemanggilannya ada pada fungsi main yang membaca kedua sumber, memvalidasi, memuat, menguji, lalu mencetak ringkasan. Skrip mencetak ringkasan baris per tabel yang berfungsi sebagai laporan singkat setiap kali pipeline selesai.

Berkas ini sengaja tetap satu program pendek tanpa konfigurasi tambahan, supaya seluruh alurnya dapat dibaca dari atas ke bawah dalam sekali duduk. Fungsi dipisah per tahap dan diberi docstring satu baris, sehingga pemetaan dari rencana ke kode tetap satu banding satu. Angka pada ringkasan diambil dari query count(*) atas tabel yang baru dimuat, bukan dari variabel penghitung yang dikelola manual.

python pipeline_proyek.py 
"""Pipeline batch mini untuk proyek peminjaman perpustakaan.

Tahapan: ekstraksi, validasi, transformasi SQL, dan muat ke basis data
DuckDB. Jalankan dari dalam direktori topik: python3 pipeline_proyek.py
"""

import json
from pathlib import Path

import duckdb

BERKAS_CSV = Path("peminjaman.csv")
BERKAS_JSON = Path("anggota.json")
BERKAS_DB = Path("peminjaman_proyek.duckdb")

TRANSFORMASI_SQL = """
CREATE OR REPLACE TABLE dim_anggota AS
SELECT id_anggota, nama, kelompok
FROM anggota_mentah;

CREATE OR REPLACE TABLE dim_buku AS
SELECT
    ROW_NUMBER() OVER (ORDER BY judul_buku) AS id_buku,
    judul_buku,
    kategori
FROM (SELECT DISTINCT judul_buku, kategori FROM peminjaman_mentah);

CREATE OR REPLACE TABLE fakta_peminjaman AS
SELECT
    p.id_pinjam,
    p.tanggal_pinjam,
    p.tanggal_kembali,
    date_diff('day', p.tanggal_pinjam, p.tanggal_kembali) AS durasi_hari,
    b.id_buku,
    a.id_anggota
FROM peminjaman_mentah p
JOIN dim_buku b USING (judul_buku, kategori)
JOIN dim_anggota a USING (id_anggota);
"""


def ekstrak_csv() -> list[dict]:
    """Baca berkas CSV sumber menjadi daftar kamus."""
    with BERKAS_CSV.open(encoding="utf-8") as berkas:
        header = berkas.readline().strip().split(",")
        return [
            dict(zip(header, baris.strip().split(",")))
            for baris in berkas
            if baris.strip()
        ]


def ekstrak_json() -> list[dict]:
    """Baca berkas JSON sumber menjadi daftar kamus."""
    with BERKAS_JSON.open(encoding="utf-8") as berkas:
        return json.load(berkas)


def validasi(baris: list[dict], anggota: list[dict]) -> None:
    """Aserasi sederhana atas data hasil ekstraksi."""
    assert baris, "berkas peminjaman.csv kosong"
    assert anggota, "berkas anggota.json kosong"
    terdaftar = {a["id_anggota"] for a in anggota}
    id_pinjam = []
    for p in baris:
        assert p["id_anggota"] in terdaftar, (
            f"id anggota tidak terdaftar: {p['id_anggota']}"
        )
        if p["tanggal_kembali"]:
            assert p["tanggal_kembali"] >= p["tanggal_pinjam"], (
                f"urutan tanggal terbalik pada {p['id_pinjam']}"
            )
        id_pinjam.append(p["id_pinjam"])
    assert len(id_pinjam) == len(set(id_pinjam)), "id_pinjam ganda pada sumber"


def muat_dan_transformasi(baris: list[dict], anggota: list[dict]) -> duckdb.DuckDBPyConnection:
    """Muat data ke tabel staging lalu bangun tabel dimensi dan fakta."""
    if BERKAS_DB.exists():
        BERKAS_DB.unlink()
    con = duckdb.connect(str(BERKAS_DB))
    con.execute(
        "CREATE TABLE peminjaman_mentah ("
        "id_pinjam VARCHAR, tanggal_pinjam DATE, tanggal_kembali DATE, "
        "id_anggota VARCHAR, judul_buku VARCHAR, kategori VARCHAR)"
    )
    data = [
        (
            p["id_pinjam"],
            p["tanggal_pinjam"],
            p["tanggal_kembali"] or None,
            p["id_anggota"],
            p["judul_buku"],
            p["kategori"],
        )
        for p in baris
    ]
    con.executemany("INSERT INTO peminjaman_mentah VALUES (?, ?, ?, ?, ?, ?)", data)
    con.execute(
        "CREATE TABLE anggota_mentah (id_anggota VARCHAR, nama VARCHAR, kelompok VARCHAR)"
    )
    con.executemany(
        "INSERT INTO anggota_mentah VALUES (?, ?, ?)",
        [(a["id_anggota"], a["nama"], a["kelompok"]) for a in anggota],
    )
    for pernyataan in TRANSFORMASI_SQL.strip().split(";"):
        if pernyataan.strip():
            con.execute(pernyataan)
    return con


def uji_muatan(con: duckdb.DuckDBPyConnection, jumlah_sumber: int) -> None:
    """Pengujian pascamuat: jumlah baris fakta harus sama dengan sumber."""
    n = con.execute("SELECT count(*) FROM fakta_peminjaman").fetchone()[0]
    assert n == jumlah_sumber, (
        f"jumlah fakta ({n}) tidak sama dengan sumber ({jumlah_sumber})"
    )


def ringkasan(con: duckdb.DuckDBPyConnection) -> str:
    """Kumpulkan ringkasan baris per tabel untuk laporan."""
    hasil = []
    tabel_daftar = (
        "peminjaman_mentah",
        "anggota_mentah",
        "dim_buku",
        "dim_anggota",
        "fakta_peminjaman",
    )
    for tabel in tabel_daftar:
        n = con.execute(f"SELECT count(*) FROM {tabel}").fetchone()[0]
        hasil.append(f"{tabel:<20}: {n} baris")
    rata = con.execute(
        "SELECT round(avg(durasi_hari), 1) FROM fakta_peminjaman "
        "WHERE durasi_hari IS NOT NULL"
    ).fetchone()[0]
    hasil.append(f"{'durasi rata-rata':<20}: {rata} hari")
    terbuka = con.execute(
        "SELECT count(*) FROM fakta_peminjaman WHERE tanggal_kembali IS NULL"
    ).fetchone()[0]
    hasil.append(f"{'pinjaman terbuka':<20}: {terbuka} baris")
    return "\n".join(hasil)


def main() -> None:
    baris = ekstrak_csv()
    anggota = ekstrak_json()
    validasi(baris, anggota)
    print(f"ekstraksi lulus: {len(baris)} peminjaman, {len(anggota)} anggota")
    con = muat_dan_transformasi(baris, anggota)
    uji_muatan(con, len(baris))
    print("pengujian pascamuat lulus")
    print(ringkasan(con))
    con.close()


if __name__ == "__main__":
    main()
text Keluaran 
ekstraksi lulus: 12 peminjaman, 5 anggota
pengujian pascamuat lulus
peminjaman_mentah   : 12 baris
anggota_mentah      : 5 baris
dim_buku            : 9 baris
dim_anggota         : 5 baris
fakta_peminjaman    : 12 baris
durasi rata-rata    : 3.1 hari
pinjaman terbuka    : 3 baris

Ringkasan menunjukkan dua tabel staging memuat seluruh baris sumber, tiga tabel target terbentuk, dan pengujian pascamuat lulus karena jumlah fakta sama dengan jumlah sumber. Nilai durasi rata-rata 3.1 hari dihitung dari pinjaman yang sudah memiliki tanggal kembali. Tiga pinjaman yang masih terbuka berasal dari tiga baris sumber dengan tanggal_kembali kosong. Tidak ada satu pun angka yang diketik manual karena ringkasan selalu dihitung ulang dari basis data yang baru dimuat.

Menjadwalkan dengan Airflow#

Pipeline yang benar saat dijalankan manual tetap perlu dijadwalkan agar SLA pukul 07:00 terjaga tanpa keterlibatan manusia. Berkas dag_peminjaman.py membungkus perintah pipeline dalam BashOperator dan menjadwalkannya pada pukul 06:30 setiap hari, 30 menit sebelum tenggat muat. Parameter catchup bernilai False sehingga run yang terlewat karena instalasi baru tidak diulang mundur. Satu run berarti satu eksekusi DAG.

python dag_peminjaman.py 
# tidak-dijalankan: butuh runtime Airflow yang tidak tersedia di sandbox
# DAG peminjaman perpustakaan: menjadwalkan pipeline_proyek.py setiap hari
# pukul 06:30, 30 menit sebelum tenggat muat SLA pukul 07:00.
# Tempatkan berkas ini di folder dags/ instalasi Airflow, bersama
# pipeline_proyek.py serta berkas sumbernya.

import pendulum
from airflow import DAG
from airflow.providers.standard.operators.bash import BashOperator

with DAG(
    dag_id="peminjaman_perpustakaan",
    schedule="30 6 * * *",
    start_date=pendulum.datetime(2026, 9, 25, tz="Asia/Jakarta"),
    catchup=False,
    tags=["proyek-akhir"],
) as dag:
    jalankan_pipeline = BashOperator(
        task_id="jalankan_pipeline",
        bash_command="python3 pipeline_proyek.py",
    )

Berkas DAG tidak dijalankan pada lingkungan latihan karena butuh runtime Airflow, tetapi strukturnya sengaja dibuat minimal supaya mudah dipindahkan ke folder dags/. Jalur impor mengikuti konvensi Airflow 3, yaitu paket standard provider untuk BashOperator. Apabila jadwal berubah, misalnya menjadi dua kali sehari, yang disunting hanya ekspresi jadwalnya tanpa menyentuh pipeline.

Jadwal 06:30 memberi ruang 30 menit sebelum tenggat, cukup untuk satu kali percobaan ulang tanpa menembus batas SLA. Berkas DAG berada terpisah dari pipeline karena keduanya berubah dengan ritme yang berbeda, yaitu logika pemrosesan lebih jarang berubah daripada penjadwalannya. Kata kunci with membuat DAG terikat rapi pada bloknya, sehingga task yang dideklarasikan di dalamnya otomatis menjadi bagian DAG tersebut.

Pengujian dan eksekusi ulang#

Pipeline yang dijadwalkan pasti dieksekusi ulang, baik oleh percobaan ulang orkestrator maupun oleh manusia setelah kegagalan. Karena itu pipeline dibuat idempoten, yaitu hasilnya tidak bergantung pada berapa kali ia dijalankan. Basis data target dihapus dan dibuat ulang di awal setiap eksekusi, sehingga tidak ada baris yang terduplikasi dan keluarannya selalu sama.

Pengujian pada proyek ini bertingkat, yaitu aserasi sebelum transformasi, pengujian pascamuat yang membandingkan jumlah fakta dengan jumlah sumber, dan cek eksekusi ulang yang membandingkan keluaran dua eksekusi. Ketiganya berjalan di dalam pipeline sendiri tanpa alat tambahan, sehingga murah dipelihara dan tetap berjalan di lingkungan mana pun.

bash 
# eksekusi pertama, basis data dibuat ulang dari awal
python3 pipeline_proyek.py
# eksekusi kedua, keluaran harus sama persis dengan eksekusi pertama
python3 pipeline_proyek.py
text Keluaran 
ekstraksi lulus: 12 peminjaman, 5 anggota
pengujian pascamuat lulus
peminjaman_mentah   : 12 baris
anggota_mentah      : 5 baris
dim_buku            : 9 baris
dim_anggota         : 5 baris
fakta_peminjaman    : 12 baris
durasi rata-rata    : 3.1 hari
pinjaman terbuka    : 3 baris
ekstraksi lulus: 12 peminjaman, 5 anggota
pengujian pascamuat lulus
peminjaman_mentah   : 12 baris
anggota_mentah      : 5 baris
dim_buku            : 9 baris
dim_anggota         : 5 baris
fakta_peminjaman    : 12 baris
durasi rata-rata    : 3.1 hari
pinjaman terbuka    : 3 baris

Kedua eksekusi mencetak ringkasan yang identik, dan itulah bukti idempotensi yang paling sederhana. Kebiasaan mengecek idempotensi dengan eksekusi ganda murah dilakukan dan menangkap banyak bug muat sebelum pipeline bertemu data yang lebih besar. Perilaku yang sama juga menjadi dasar percobaan ulang otomatis oleh orkestrator.

Kesalahan umum#

Pembangunan pipeline menampakkan kesalahan desain sekaligus memunculkan kesalahan barunya. Empat pola berikut paling sering menghambat proyek akhir. Semuanya dicegah dengan kebiasaan yang sama, yaitu menguji pipeline terhadap data nyata lebih awal.

  • Aserasi yang terlalu longgar. Pemeriksaan yang hanya menguji kekosongan berkas membiarkan id tak dikenal dan tanggal terbalik lolos ke tabel fakta. Tambahkan satu aserasi untuk setiap relasi penting antar sumber.
  • Tanggal eksekusi yang tercampur ke data. Memakai tanggal hari ini pada kolom contoh membuat keluaran berubah setiap kali dijalankan. Gunakan tanggal literal pada data contoh dan simpan tanggal eksekusi hanya pada kolom khusus audit.
  • Muat yang menambah tanpa menghapus. Eksekusi ulang di atas tabel yang belum dibersihkan menduplikasi seluruh baris. Atur ulang tabel di awal eksekusi atau gunakan CREATE OR REPLACE agar pipeline tetap idempoten.
  • Validasi yang terpisah dari pipeline. Skrip pemeriksa yang berdiri sendiri mudah terlewat saat eksekusi mendesak. Tanamlah aserasi di dalam fungsi pipeline agar pemeriksaan berjalan pada setiap eksekusi.

Ringkasan#

  • Pipeline mini berjalan utuh dari CSV dan JSON sampai tiga tabel target pada satu berkas DuckDB, dengan ringkasan baris per tabel setiap kali selesai.
  • Validasi berupa empat aserasi yang menghentikan pipeline sebelum data kotor masuk transformasi.
  • DAG Airflow menjadwalkan perintah pipeline pukul 06:30 dengan catchup False untuk memenuhi SLA pukul 07:00.
  • Basis data diatur ulang di awal eksekusi sehingga pipeline idempoten dan keluarannya deterministik.

Referensi#