Belajar Data Engineer - Python untuk Data Engineering

#Pendahuluan

Di episode 4 kita mendesain schema yang rapi — sekarang kalian perlu mengisi warehouse itu dengan data dari dunia nyata. Dan di dunia nyata, data datang dari API, file, dan database dengan segala keanehannya: kolom kosong, format tanggal berubah-ubah, karakter aneh, sampai respons yang gagal di tengah jalan.

Episode ini membangun sisi Python dari toolkit data engineer: pandas/polars untuk transformasi, requests untuk menarik data API, error handling agar script tidak mati diam-diam, dan testing untuk memastikan kode data tidak mudah rusak. Semua dikemas dalam satu praktik: script ekstraksi data API yang layak produksi.

#Pandas vs Polars

Pandas adalah standar lama untuk manipulasi data tabular di Python. Polars adalah pendatang baru yang lebih cepat dan hemat memori karena dibangun di atas Rust dan memakai lazy execution serta model multi-thread. Di tahun 2026, polars adalah pilihan yang lebih baik untuk data engineering — tapi memahami pandas tetap berguna karena ekosistemnya luas.

Transformasi yang sama di kedua library:

import pandas as pd
 
df = pd.read_csv("orders.csv")
df["amount"] = pd.to_numeric(df["amount"], errors="coerce")
summary = df[df["status"] == "completed"] \
    .groupby("customer_id")["amount"] \
    .agg(["count", "sum"]) \
    .reset_index()
print(summary.head())

import polars as pl
 
df = pl.read_csv("orders.csv")
summary = (
    df.filter(pl.col("status") == "completed")
    .group_by("customer_id")
    .agg(pl.col("amount").count().alias("count"),
         pl.col("amount").sum().alias("sum"))
    .sort("sum", descending=True)
)
print(summary.head())

Perhatikan pl.col("amount").count().alias("count") — polars menuntut eksplisit: setiap kolom hasil harus diberi nama dengan .alias(). Eksplisit ini yang membuat pipeline polars lebih mudah didebug.

#Mengambil Data dari API dengan requests

API adalah sumber data paling umum di dunia nyata. Kunci menulis ingestion yang andal: pagination, rate limiting, dan timeout.

import time
import requests
import polars as pl
 
BASE = "https://api.example.com"
TOKEN = "YOUR_TOKEN"
HEADERS = {"Authorization": f"Bearer {TOKEN}"}
 
def fetch_orders_page(page: int) -> list[dict]:
    resp = requests.get(
        f"{BASE}/orders",
        headers=HEADERS,
        params={"page": page, "limit": 100},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()["data"]
 
def fetch_all_orders() -> pl.DataFrame:
    rows: list[dict] = []
    page = 1
    while True:
        data = fetch_orders_page(page)
        if not data:
            break
        rows.extend(data)
        page += 1
        time.sleep(0.5)  # rate limiting: hormati batas API
    return pl.DataFrame(rows)

Poin penting:

• timeout=30 — tanpa timeout, script bisa menggantung selamanya saat API lambat.
• resp.raise_for_status() — mengubah HTTP error menjadi exception yang bisa ditangani, bukan hasil diam-diam.
• time.sleep(0.5) — mencegah kalian di-ban karena request terlalu cepat.

Pola fetch_page (I/O murni) yang dipisah dari transformasi adalah cikal bakal script ekstraksi produksi: fungsi I/O bisa diuji terpisah, dan transformasi menjadi fungsi murni yang mudah ditest — kita rangkai lengkap di bagian testing.

#Error Handling yang Benar

Kode data produksi tidak boleh "menyerah" begitu saja, tapi juga tidak boleh menyembunyikan kegagalan. Pola yang benar: tahu kapan harus retry, kapan harus fail.

import time
from requests.exceptions import RequestException
 
def fetch_with_retry(url: str, headers: dict, attempts: int = 3) -> dict:
    for attempt in range(attempts):
        try:
            resp = requests.get(url, headers=headers, timeout=30)
            resp.raise_for_status()
            return resp.json()
        except RequestException as e:
            if attempt == attempts - 1:
                raise  # gagal total — biarkan orkestrator tahu
            time.sleep(2 ** attempt)  # exponential backoff: 2s, 4s

Logika di baliknya: kesalahan jaringan dan 5xx bersifat sementara — layak retry dengan backoff. Sedangkan 4xx (misalnya 401 unauthorized) adalah kesalahan permanen — retry tidak akan membantu, lebih baik gagal cepat dengan pesan yang jelas.

#Testing Kode Data

Script data sering dianggap "sekali pakai" — ini mitos yang mahal. Pipeline yang berjalan tiap malam adalah software dan layak diuji. Dengan pytest, fokus uji pada transformasi murni (fungsi tanpa I/O) karena itu yang paling mudah dan paling rentan salah:

from transform import standardize_orders
 
def test_standardize_orders_snapshot():
    raw = [
        {"id": "1", "amount": "Rp 25.000", "date": "01-08-2026"},
        {"id": "2", "amount": "Rp 30.000", "date": "02-08-2026"},
    ]
    result = standardize_orders(raw)
    assert result["amount"].to_list() == [25000, 30000]
    assert result["date"].to_list() == ["2026-08-01", "2026-08-02"]
 
def test_standardize_orders_bad_row_skipped():
    raw = [{"id": "x", "amount": "tidak valid", "date": "?"}]
    result = standardize_orders(raw)
    assert len(result) == 0

Test di atas menangkap dua hal penting: snapshot test memastikan output stabil, dan test kasus abnormal memastikan baris rusak ditangani dengan benar (misalnya dilewati, bukan menghentikan seluruh pipeline).

#Kesalahan Umum (Common Pitfalls)

1. Tanpa timeout. API yang menggantung membuat pipeline stuck diam-diam. Selalu set timeout.

2. except Exception: pass. Menyembunyikan error dan membuat data kosong lolos tanpa ketahuan. Fail loud atau log lengkap.

3. Menjalankan transformasi langsung di satu blok raksasa. Pecah jadi fungsi murni agar bisa diuji dan dipakai ulang.

4. Mengabaikan rate limiting. Request terlalu cepat ke API publik = di-ban atau terkena 429. Hormati Retry-After.

#Penutup

Di episode 5 ini kalian telah mengubah Python menjadi senjata produksi data engineering:

• pandas vs polars: polars lebih cepat & hemat memori untuk data engineering; pandas untuk ekosistem lama.
• requests dengan timeout, raise_for_status, dan rate limiting untuk ingestion API yang andal.
• Error handling: retry dengan exponential backoff untuk error sementara, fail cepat untuk error permanen.
• Testing dengan pytest memastikan transformasi tidak rusak diam-diam.

Di episode 6 selanjutnya kita akan membahas keputusan arsitektur paling fundamental dalam pipeline: ETL vs ELT & tools — perbedaan konsep, tool modern seperti Airbyte, Fivetran, dbt, dan Stitch, trade-off masing-masing, plus praktik membangun pipeline ELT sederhana. Sampai jumpa di episode 6!