Prinsip zona raw pada data lake adalah menyimpan data persis seperti diterima dari sumber, sebelum transformasi apa pun. Bila validasi Bagian 6 keliru atau analisis di hilir menemukan kejanggalan, kita selalu bisa memutar ulang proses dari data asli. Format pilihan kita adalah JSON Lines: satu objek JSON utuh per baris teks, sehingga berkas bisa dibaca mengalir, dihitung dengan wc -l, dan diproses baris demi baris tanpa memuat semuanya ke memori. Pada bagian ini kita membangun penyimpanan di src/storage.py dan menjalankan pipeline lengkap untuk pertama kalinya.

Penulisan sengaja memakai mode append karena zona raw bersifat tambah-saja:

python 
def write_jsonl(path, rows):
    """Tambahkan ``rows`` ke akhir berkas JSONL; kembalikan jumlah baris.

    Mode tulis selalu ``append`` karena zona raw bersifat tambah-saja
    (append-only): run lama tidak pernah ditimpa.
    """
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as berkas:
        for baris in rows:
            berkas.write(json.dumps(baris, ensure_ascii=False) + "\n")
    return len(rows)

Mari kita telaah baris-baris kode di atas:

  1. Baris 41 (mkdir(parents=True, exist_ok=True)): Direktori data/raw/ dibuat otomatis pada run pertama, sehingga proyek baru langsung jalan tanpa persiapan manual.
  2. Baris 42 (path.open("a", ...)): Mode a menambahkan di akhir berkas; bila berkas belum ada, Python membuat berkas baru, sehingga satu fungsi menangani dua kondisi.
  3. Baris 44 (json.dumps(...) + "\n"): Setiap baris adalah dokumen JSON lengkap diakhiri newline; inilah seluruh kontrak format JSON Lines.
  4. Baris 45 (return len(rows)): Jumlah baris yang benar-benar ditulis dikembalikan agar pemanggil bisa mencocokkannya dengan jumlah baris valid.

Fungsi read_jsonl membaca kebalikannya dengan pemeriksaan ketat: baris yang bukan JSON valid memunculkan ValueError yang menyebut nomor baris persisnya, sehingga berkas rusak mudah diperbaiki.

Tata Nama Berkas Run #

Setiap eksekusi pipeline mendapat run_id berformat UTC seperti 20261017T074428Z, dan berkas run dinamai <dataset>-<run_id>.jsonl. Dua run pada detik yang sama berpotensi menghasilkan nama sama, sehingga alokasi nama menambahkan sufiks nomor:

python 
def allocate_raw_path(raw_dir, dataset, rid):
    """Path run yang belum dipakai; tambahkan ``-2``, ``-3``, ... bila bentrok.

    Dua run yang dimulai pada detik yang sama menghasilkan ``run_id`` sama.
    Sufiks nomor memastikan berkas run sebelumnya tidak pernah tertimpa.
    """
    kandidat = raw_path(raw_dir, dataset, rid)
    nomor = 2
    while kandidat.exists():
        kandidat = raw_path(raw_dir, dataset, f"{rid}-{nomor}")
        nomor += 1
    return kandidat

Pemeriksaan while kandidat.exists() berjalan sebelum penulisan, jadi run kedua pada detik yang sama menulis ke pesanan-...-2.jsonl, bukan menimpa run pertama. Sifat tidak pernah menimpa inilah yang membuat zona raw aman untuk diputar ulang kapan pun.

Run Pertama Pipeline #

Semua komponen kini dirangkai oleh src/pipeline.py: mengambil seluruh halaman, melepas duplikat, memvalidasi, menulis JSONL, lalu menyimpan state kursor. Perintah berikut menjalankan pipeline mode full dengan fake API lokal sebagai sumber:

bash 
python3 -m src.pipeline --mode full --demo

Output yang Dihasilkan:

text 
INFO ingestion.pipeline: fake API demo berjalan di http://127.0.0.1:44873
INFO ingestion.pipeline: run 20261017T074428Z mode=full kursor=-
INFO ingestion.pagination: halaman 1/3: 50 baris
INFO ingestion.pagination: halaman 2/3: 50 baris
INFO ingestion.pagination: halaman 3/3: 37 baris
INFO ingestion.pipeline: run 20261017T074428Z selesai: 137 diambil, 137 ditulis, kursor=2024-01-05T16:16:00+00:00
=== Ringkasan Run ===
mode           : full
run_id         : 20261017T074428Z
baris_diambil  : 137
baris_duplikat : 0
baris_valid    : 137
baris_dilewati : 0
output         : data/raw/pesanan-20261017T074428Z.jsonl
kursor         : 2024-01-05T16:16:00+00:00

Log INFO memperlihatkan tiga halaman dari Bagian 3, dan ringkasan mencatat 137 baris valid tanpa satu pun yang dilewati. Sekarang kita periksa berkas yang dihasilkan:

bash 
wc -l data/raw/pesanan-20261017T074428Z.jsonl

Output yang Dihasilkan:

text 
137 data/raw/pesanan-20261017T074428Z.jsonl
bash 
head -n 1 data/raw/pesanan-20261017T074428Z.jsonl | python3 -m json.tool

Output yang Dihasilkan:

text 
{
    "id": 1,
    "pelanggan": "Budi Santoso",
    "email": "[email protected]",
    "total": 1151500,
    "status": "dibayar",
    "created_at": "2024-01-01T08:20:00+00:00",
    "updated_at": "2024-01-01T08:49:00+00:00"
}

Baris pertama berkas identik dengan respons API pada Bagian 1, bukti bahwa zona raw menyimpan data mentah tanpa perubahan. Karena formatnya satu objek per baris, alat baris perintah biasa langsung bisa bekerja, misalnya menghitung pesanan per status:

bash 
jq -r .status data/raw/pesanan-*.jsonl | sort | uniq -c

Output yang Dihasilkan:

text 
     26 baru
     27 dibatalkan
     19 dibayar
     26 dikirim
     39 selesai

Uji Penyimpanan #

Tes di tests/test_storage.py memakai direktori sementara tmp_path dari pytest, sehingga tes tidak pernah menyentuh data/raw/ milik proyek. Pasangan tes berikut memeriksa penulisan bolak-balik dan sifat tambah-saja; tes lainnya memastikan nama run unik lewat sufiks -2 saat bentrok:

python 
def test_tulis_lalu_baca_bolak_balik(tmp_path):
    path = tmp_path / "pesanan.jsonl"
    assert write_jsonl(path, [{"id": 1}, {"id": 2}]) == 2
    assert read_jsonl(path) == [{"id": 1}, {"id": 2}]


def test_write_jsonl_menambah_bukan_menimpa(tmp_path):
    path = tmp_path / "pesanan.jsonl"
    write_jsonl(path, [{"id": 1}])
    write_jsonl(path, [{"id": 2}])
    assert len(read_jsonl(path)) == 2
bash 
python3 -m pytest tests/test_storage.py -q

Output yang Dihasilkan:

text 
.......                                                                  [100%]
7 passed in 0.13s

Zona raw kini berisi data mentah yang teruji, sesuai konsep zona pada chapter Zona Data Raw dan Curated dan format yang dibahas pada chapter JSON Lines; spesifikasi formatnya terdokumentasi di jsonlines.org . Namun setiap run masih mengambil seluruh 137 pesanan; bayangkan dataset jutaan baris yang diunduh ulang setiap jam. Bagian berikutnya memotong pekerjaan tersebut dengan mode incremental berbasis kursor updated_at.