Data Pipeline dengan Python — Data Engineering

Membangun Data Pipeline dengan Python Data pipeline adalah serangkaian langkah otomatis yang mengambil data dari sumber, mentransformasi, dan memuatnya ke…

Data pipeline dengan Python adalah satu script yang menjalankan extract, transform, dan load untuk satu tanggal, mencatat apa yang terjadi, dan aman dijalankan ulang. Bayangin buku kas toko yang diisi tiap malam oleh siapa pun yang jaga. Caranya ditulis di kertas yang ditempel di dinding: ambil struk hari itu, buang struk dobel, tulis totalnya di halaman tanggal itu. Kalau ternyata salah, halaman tanggal itu dihapus dan ditulis ulang, bukan ditambah di bawahnya. Kertas di dinding itu script-nya, halaman per tanggal itu partisi di warehouse.

🧪 Tebakan cepat: script di pelajaran ini dijalankan dua kali untuk 1 Oktober. Baris terakhir log run kedua berbunyi load 2026-10-01: 2 baris, penjualan sekarang ... baris. Angkanya 2 atau 4? Tebak dulu, jawabannya ada di bagian "Load: hapus tanggal itu, lalu isi".

Bentuk pipeline-nya

Langkah Fungsi di pipeline.py Di Toko Buku
Extract extract(tanggal) SQL ke database aplikasi, pesanan satu hari
Transform transform(df) pandas: buang baris dobel, kota kosong, harga katalog, minggu
Load load(df, tanggal) tabel penjualan di warehouse DuckDB gudang.duckdb

Database aplikasi Toko Buku itu MySQL. Supaya bisa jalan di laptop tanpa server, di sini sumbernya file SQLite toko.db berisi tabel yang sama persis. Di produksi kamu cukup mengganti cara konek (misalnya lewat PyMySQL atau SQLAlchemy); query-nya tetap. Library-nya dipasang dengan pip install pandas duckdb (pandas 3.0.6, DuckDB 1.5.6). Script-nya satu file, dibahas per bagian.

Config dari environment

import argparse
import logging
import os
import sqlite3
import time
from datetime import date, timedelta

import duckdb
import pandas as pd

# Konfigurasi dari environment variable, bukan ditulis di kode
SUMBER_DB = os.environ.get("SUMBER_DB", "toko.db")
GUDANG_DB = os.environ.get("GUDANG_DB", "gudang.duckdb")
PERCOBAAN = int(os.environ.get("PERCOBAAN", "3"))

log = logging.getLogger("pipeline")

Extract dengan retry

EXTRACT_SQL = """
SELECT oi.id AS item_id, oi.order_id, oi.book_id, o.created_at,
       c.city, b.title, oi.qty, oi.price AS harga_bayar, b.price AS harga_katalog
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
JOIN customers c ON c.id = o.customer_id
JOIN books b ON b.id = oi.book_id
WHERE o.created_at >= ? AND o.created_at < ?
"""


def extract(tanggal: date) -> pd.DataFrame:
    besok = tanggal + timedelta(days=1)
    for ke in range(1, PERCOBAAN + 1):
        try:
            con = sqlite3.connect(f"file:{SUMBER_DB}?mode=ro", uri=True)
            try:
                df = pd.read_sql_query(EXTRACT_SQL, con, params=(tanggal.isoformat(), besok.isoformat()))
            finally:
                con.close()
            log.info("extract %s: %d baris", tanggal, len(df))
            return df
        except sqlite3.OperationalError as e:
            if ke == PERCOBAAN:
                raise
            tunggu = 2 ** (ke - 1)
            log.warning("extract gagal (%s), percobaan %d dari %d, coba lagi %d detik lagi", e, ke, PERCOBAAN, tunggu)
            time.sleep(tunggu)

Transform dengan pandas

def transform(df: pd.DataFrame) -> pd.DataFrame:
    for row in df[df["harga_bayar"] != df["harga_katalog"]].itertuples():
        log.warning("item %d: harga bayar %d, katalog %d", row.item_id, row.harga_bayar, row.harga_katalog)
    bersih = df.drop_duplicates(subset=["order_id", "book_id"])
    if len(bersih) < len(df):
        log.warning("%d baris dobel dibuang", len(df) - len(bersih))
    waktu = pd.to_datetime(bersih["created_at"])
    return pd.DataFrame({
        "order_id": bersih["order_id"],
        "book_id": bersih["book_id"],
        "tanggal": waktu.dt.date,
        "minggu": waktu.dt.to_period("W").dt.start_time.dt.date,
        "kota": bersih["city"].fillna("Tidak diketahui"),
        "judul": bersih["title"],
        "qty": bersih["qty"],
        "harga": bersih["harga_katalog"],
    })

Ini langkah dari pelajaran pandas, ditambah dua pengecekan yang mencatat WARNING: harga bayar yang beda dengan katalog (item 16), dan baris dobel (item 9). Fungsinya cuma menerima DataFrame dan mengembalikan DataFrame, jadi gampang dites tanpa database.

Pengecekan seperti ini bisa dibuat lebih rapi dengan library validasi: Pydantic memeriksa tiap record sesuai schema (tipe dan aturan per kolom), Great Expectations memeriksa satu tabel dengan aturan deklaratif seperti "harga harus antara 1 dan 500.000". Detailnya di pelajaran Data Quality & Validation.

Load: hapus tanggal itu, lalu isi

def load(df: pd.DataFrame, tanggal: date) -> None:
    con = duckdb.connect(GUDANG_DB)
    try:
        con.execute("""
            CREATE TABLE IF NOT EXISTS penjualan (
                order_id INTEGER, book_id INTEGER, tanggal DATE, minggu DATE,
                kota VARCHAR, judul VARCHAR, qty INTEGER, harga INTEGER)
        """)
        con.register("baru", df)
        con.begin()
        con.execute("DELETE FROM penjualan WHERE tanggal = ?", [tanggal])
        con.execute("INSERT INTO penjualan SELECT * FROM baru")
        con.commit()
        total = con.execute("SELECT COUNT(*) FROM penjualan").fetchone()[0]
        log.info("load %s: %d baris, penjualan sekarang %d baris", tanggal, len(df), total)
    finally:
        con.close()

CLI dan backfill

def jalankan(tanggal: date) -> None:
    load(transform(extract(tanggal)), tanggal)


def main() -> None:
    parser = argparse.ArgumentParser(description="Load harian penjualan Toko Buku")
    parser.add_argument("tanggal", type=date.fromisoformat, help="tanggal pesanan, YYYY-MM-DD")
    parser.add_argument("--sampai", type=date.fromisoformat, help="tanggal terakhir, untuk backfill")
    args = parser.parse_args()
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", datefmt="%H:%M:%S")

    hari = args.tanggal
    while hari <= (args.sampai or args.tanggal):
        jalankan(hari)
        hari += timedelta(days=1)


if __name__ == "__main__":
    main()

Hasil menjalankannya:

$ python pipeline.py 2026-10-01
21:24:23 INFO extract 2026-10-01: 3 baris
21:24:23 WARNING 1 baris dobel dibuang
21:24:23 INFO load 2026-10-01: 2 baris, penjualan sekarang 2 baris
$ python pipeline.py 2026-10-01
21:24:24 INFO extract 2026-10-01: 3 baris
21:24:24 WARNING 1 baris dobel dibuang
21:24:24 INFO load 2026-10-01: 2 baris, penjualan sekarang 2 baris
$ python pipeline.py 2026-09-28 --sampai 2026-10-09
21:24:25 INFO extract 2026-09-28: 3 baris
21:24:25 INFO load 2026-09-28: 3 baris, penjualan sekarang 5 baris
...
21:24:25 INFO extract 2026-10-03: 0 baris
21:24:25 INFO load 2026-10-03: 0 baris, penjualan sekarang 9 baris
...
21:24:25 INFO extract 2026-10-07: 1 baris
21:24:25 WARNING item 16: harga bayar 950000, katalog 95000
21:24:25 INFO load 2026-10-07: 1 baris, penjualan sekarang 15 baris
...
21:24:26 INFO load 2026-10-09: 1 baris, penjualan sekarang 18 baris

Run kedua tetap 2 baris. Itu jawaban tebakannya: DELETE membuang hasil run pertama sebelum INSERT. Backfill 12 tanggal berakhir di 18 baris, 21 buku, Rp 2.214.000, sama dengan semua pelajaran sebelumnya; 1 Oktober ikut di-load lagi tanpa dobel. 3 Oktober gak ada pesanan: 0 baris itu normal, bukan error. Kalau DELETE-nya dihapus dari script, run kedua 1 Oktober mencatat penjualan sekarang 4 baris.

Kalau sumbernya gak bisa dibuka:

$ SUMBER_DB=salah.db python pipeline.py 2026-10-02
21:24:26 WARNING extract gagal (unable to open database file), percobaan 1 dari 3, coba lagi 1 detik lagi
21:24:27 WARNING extract gagal (unable to open database file), percobaan 2 dari 3, coba lagi 2 detik lagi
Traceback (most recent call last):
  ...
sqlite3.OperationalError: unable to open database file
$ echo $?
1

Exit code 1 itu penting: cron atau Airflow (pelajaran berikutnya) tahu run-nya gagal dan bisa memberi tahu orang.

⚠️ Jebakan yang sering ketemu

TL;DR: Pipeline Python untuk Toko Buku adalah satu pipeline.py: extract dengan SQL berparameter (?) ke database aplikasi untuk satu tanggal, transform dengan pandas (buang dobel, isi kota kosong, harga katalog, minggu), load ke penjualan di DuckDB dengan DELETE tanggal itu lalu INSERT dalam satu transaksi, jadi run ulang tetap 2 baris untuk 1 Oktober. Config datang dari environment variable (SUMBER_DB, GUDANG_DB), tiap langkah dicatat logging dengan jam dan level, extract di-retry dengan backoff 1 dan 2 detik lalu gagal dengan exit code 1, dan tanggalnya argumen CLI sehingga backfill 2026-09-28 --sampai 2026-10-09 mengisi 18 baris. Validasi bisa dirapikan dengan Pydantic (per record) atau Great Expectations (per tabel).

Yang akan kamu pelajari