Bagian 2: Mengambil Data Sumber

Setiap malam, sistem kasir TokoKita menutup hari dengan menuliskan satu berkas CSV yang berisi seluruh pesanan hari tersebut. Berkas-berkas itu menumpuk di zona raw dengan pola nama yang tetap, dan pipeline kita harus menemukan partisi yang tepat untuk satu tanggal, membacanya tanpa mengubah apa pun, lalu menolaknya dengan galat yang jelas bila bentuknya tidak sesuai kontrak. Tahap extract adalah pintu masuk tunggal; kesalahan di sini menjalar ke seluruh pipeline. Karena itu kita membiarkan data mentah tetap mentah: spasi berlebih, harga berformat Indonesia, dan status huruf besar diterima apa adanya, lalu dibersihkan pada tahap berikutnya.

Pola Nama Partisi #

Kontrak pertama kita adalah penamaan berkas. Zona raw hanya berisi berkas pesanan_YYYY-MM-DD.csv, dan nama itulah satu-satunya tanda tanggal partisi:

import re
from datetime import date
from pathlib import Path

POLA_PARTISI = re.compile(r"^pesanan_(\d{4}-\d{2}-\d{2})\.csv$")


def find_partitions(raw_dir: Path) -> list[Path]:
    """Kembalikan seluruh partisi valid, terurut nama berkas."""
    if not raw_dir.is_dir():
        raise ExtractError(f"zona raw tidak ditemukan: {raw_dir}")
    return sorted(p for p in raw_dir.iterdir() if POLA_PARTISI.match(p.name))


def partition_date(path: Path) -> date:
    """Ambil tanggal dari nama partisi."""
    cocok = POLA_PARTISI.match(path.name)
    if cocok is None:
        raise ExtractError(f"nama partisi tidak dikenali: {path.name}")
    return date.fromisoformat(cocok.group(1))

Mari kita telaah potongan di atas (penomoran dihitung dari atas potongan):

1. Baris 4 (POLA_PARTISI): regex menjangkar ^ dan $ sehingga pesanan_2026-11-02.csv lolos, sedangkan pesanan_bukan-tanggal.csv tidak. Grup tangkapan (\d{4}-\d{2}-\d{2}) menyimpan tanggal.
2. Baris 8-11 (find_partitions): memindai isi zona raw lalu menyaring dengan pola; hasilnya diurutkan agar urutan pemrosesan stabil. Zona raw yang hilang langsung menjadi ExtractError.
3. Baris 14-18 (partition_date): mengubah grup tangkapan menjadi objek date bawaan Python, bukan string, sehingga pembandingan tanggal bebas kesalahan format.

Menjalankan penemuan partisi pada data proyek menghasilkan daftar berikut:

Output yang Dihasilkan:

== find_partitions(data/raw) ==
pesanan_2026-11-02.csv  -> tanggal partisi 2026-11-02
pesanan_2026-11-03.csv  -> tanggal partisi 2026-11-03
pesanan_2026-11-04.csv  -> tanggal partisi 2026-11-04

Membaca Partisi sebagai Teks Mentah #

Pembacaan memakai pandas.read_csv dengan dtype=str dan keep_default_na=False. Dua argumen ini membuat semua kolom menjadi string dan sel kosong menjadi "", bukan NaN. Konsekuensinya, partisi 2026-11-03 yang sengaja kotor terbaca seperti ini:

KOLOM_WAJIB = (
    "order_id", "customer_name", "city", "category",
    "quantity", "unit_price", "order_time", "status",
)


def read_partition(path: Path) -> pd.DataFrame:
    """Baca satu partisi sebagai teks mentah (dtype=str)."""
    if not path.exists():
        raise ExtractError(f"partisi tidak ditemukan: {path}")
    df = pd.read_csv(path, dtype=str, keep_default_na=False)
    kurang = [c for c in KOLOM_WAJIB if c not in df.columns]
    if kurang:
        raise ExtractError(
            f"{path.name}: kolom wajib hilang: {', '.join(kurang)}"
        )
    return df

Baris demi baris:

1. Baris 1-4 (KOLOM_WAJIB): delapan kolom yang wajib ada. Daftar ini adalah kontrak skema sumber; menambah kolom baru boleh, menghilangkan kolom lama tidak.
2. Baris 9-10: berkas yang tidak ada menjadi ExtractError dengan jalur lengkapnya. Inilah jalur yang menghasilkan exit code 3 pada Bagian 7.
3. Baris 11: dtype=str, keep_default_na=False menjaga nilai mentah. Tanpa ini, pandas menebak tipe dan bisa mengubah "1.250.000" atau sel kosong tanpa kita sadari.
4. Baris 12-16: pemeriksaan kolom memakai selisih daftar, bukan asumsi urutan kolom, sehingga partisi yang menulis ulang urutan kolom tetap diterima.

Tiga baris pertama partisi kotor memperlihatkan mengapa kita tidak membersihkan apa pun di tahap ini:

Output yang Dihasilkan:

== read_partition(pesanan_2026-11-03.csv) -> 3 baris pertama apa adanya ==
          order_id  customer_name       city   category quantity unit_price          order_time   status
ORD-2026-11-03-001  Budi Santoso     Jakarta elektronik        1     899000 2026-11-03 08:05:00  selesai
ORD-2026-11-03-002 Siti Rahmawati    Bandung elektronik        1  1.250.000 2026-11-03 09:40:00  SELESAI
ORD-2026-11-03-003    Andi Wijaya Yogyakarta    makanan        2      25000 2026-11-03 10:15:00 diproses

nilai mentah yang perlu dibersihkan:
  customer_name baris 1 : ' Budi Santoso '
  unit_price    baris 2 : '1.250.000'
  status        baris 2 : 'SELESAI'
  city         baris 10 : ''

Ringkasan Ekstraksi untuk Log #

Setiap tahap melaporkan angkanya sendiri. Fungsi kecil extract_report menghasilkan kamus yang dipakai log [extract] dan manifest eksekusi pada Bagian 8:

def extract_report(path: Path, df: pd.DataFrame) -> dict:
    """Ringkasan ekstraksi untuk log dan manifest."""
    return {
        "partisi": path.name,
        "baris": len(df),
        "kolom": len(df.columns),
        "ukuran": path.stat().st_size,
    }

Output yang Dihasilkan:

{'partisi': 'pesanan_2026-11-03.csv', 'baris': 10, 'kolom': 8, 'ukuran': 927}

Sepuluh baris, delapan kolom, dan 927 bytes: tiga angka yang cukup untuk mendeteksi partisi yang tiba-tiba berubah bentuk. Materi chapter Ekstraksi Data dan Ingestion dari File membahas pilihan ini lebih lanjut, dan Validasi Data saat Ingestion menjelaskan mengapa kontrak skema sebaiknya berdiri di depan pintu masuk. Pada bagian berikutnya kita membersihkan sepuluh baris mentah ini menjadi fact yang layak muat di warehouse.

Bacaan lanjut: dokumentasi pandas read_csv dan re modul Python .