Bagian 2: Mengumpulkan Data

Pipeline yang menggabungkan dua sumber harus menghormati perbedaan karakter masing-masing sumber sejak tahap baca. Ekspor pesanan berupa CSV dari basis data operasional: kolomnya tetap, tetapi nilainya bisa berformat Rp 85.000 dan tanggalnya bisa campur format. Event web berupa JSON Lines dari event tracker: tiap baris satu objek JSON, tetapi barisnya bisa terpotong karena koneksi terputus saat pengiriman. Pada bagian ini kita menulis src/collect.py yang membaca keduanya apa adanya — koersi dan pembersihan sengaja tidak dilakukan di sini supaya setiap tahap pipeline punya satu tanggung jawab. Materi pendukung ada pada chapter Ingestion dari File dan Konsep Data Ingestion.

Dua Sumber, Dua Karakter #

Kita bandingkan kedua sumber pada data/raw/ sebelum menulis kode:

Data latihan ini memang memuat kotoran yang disengaja dan realistis, karena pipeline yang hanya pernah melihat data sempurna akan langsung runtuh di produksi. Seluruh kotoran ditangani pada Bagian 3; tugas bagian ini hanya memastikan keduanya terbaca utuh tanpa ada baris yang hilang diam-diam.

Membaca Kedua Sumber #

Dua fungsi inti src/collect.py adalah baca_pesanan dan baca_events:

def baca_pesanan(jalur) -> pd.DataFrame:
    """Baca ekspor pesanan CSV sebagai teks mentah.

    Seluruh kolom dibaca sebagai string (``dtype=str``) dengan ``keep_default_na``
    dinonaktifkan, sehingga sel kosong tetap berupa string kosong dan tidak berubah
    menjadi NaN. Keputusan ini menjaga pembersihan tetap bertanggung jawab penuh
    atas interpretasi nilai.
    """
    return pd.read_csv(jalur, dtype=str, keep_default_na=False)


def baca_events(jalur):
    """Baca event web JSON Lines baris-per-baris.

    Mengembalikan pasangan ``(events, karantina)``. Baris yang gagal diuraikan
    sebagai JSON tidak dijatuhkan diam-diam, melainkan dicatat ke karantina
    baca beserta nomor baris dan isi aslinya.
    """
    events, karantina = [], []
    with open(jalur, encoding="utf-8") as berkas:
        for nomor, baris in enumerate(berkas, start=1):
            teks = baris.strip()
            if not teks:
                continue
            try:
                events.append(json.loads(teks))
            except json.JSONDecodeError:
                karantina.append(
                    {"alasan": "JSON tidak valid", "baris": nomor, "isi": teks}
                )
    return events, karantina

Mari kita telaah baris-baris kode di atas:

1. baca_pesanan baris 23 (dtype=str, keep_default_na=False): seluruh kolom CSV dibaca sebagai string dan sel kosong tetap string kosong. Tanpa opsi ini pandas akan menebak tipe dan mengubah sel kosong menjadi NaN, sehingga tahap pembersihan kehilangan informasi asli.
2. baca_events baris 33 (for nomor, baris in enumerate(...)): JSON Lines dibaca baris-per-baris dengan nomor baris disimpan, karena nomor baris adalah satu-satunya alamat baris yang rusak pada berkas JSONL.
3. baca_events baris 36-38 (if not teks: continue): baris kosong dilewati tanpa dihitung sebagai data maupun karantina.
4. baca_events baris 40-43 (try/except json.JSONDecodeError): baris yang gagal diuraikan dimasukkan ke karantina bersama alasan, nomor baris, dan isi aslinya — tidak dibuang.

Nomor baris pada uraian di atas merujuk posisi nyata pada berkas src/collect.py.

Menjalankan Pengumpulan #

Skrip berikut memprofilkan kedua sumber lalu membacanya, persis seperti yang dilakukan tahap pertama pipeline:

from src import collect

profil_pesanan = collect.profil_sumber("data/raw/pesanan.csv")
profil_events = collect.profil_sumber("data/raw/events.jsonl")

print("profil sumber:")
for profil in (profil_pesanan, profil_events):
    print(f"  {profil['nama']:14s} {profil['jumlah_baris']:4d} baris, {profil['ukuran_byte']:7d} byte")

df = collect.baca_pesanan("data/raw/pesanan.csv")
events, karantina = collect.baca_events("data/raw/events.jsonl")
print(f"pesanan terbaca : {len(df)} baris data, {len(df.columns)} kolom")
print(f"events terbaca  : {len(events)} baris JSON valid, {len(karantina)} baris rusak")
print(f"baris rusak     : nomor {karantina[0]['baris']} -> {karantina[0]['isi']!r}")

print("nilai mentah baris pertama (belum dikoersi):")
pertama = df.loc[0, ["pesanan_id", "tanggal", "harga_satuan", "diskon", "total"]]
for kunci, nilai in pertama.items():
    print(f"  {kunci:12s} = {nilai!r}")

Output yang Dihasilkan:

profil sumber:
  pesanan.csv      39 baris,    3968 byte
  events.jsonl    116 baris,   20021 byte
pesanan terbaca : 38 baris data, 12 kolom
events terbaca  : 115 baris JSON valid, 1 baris rusak
baris rusak     : nomor 116 -> '{"event_id": "EVT-0116", "waktu": "2025-01-15T1'
nilai mentah baris pertama (belum dikoersi):
  pesanan_id   = 'ORD-2101'
  tanggal      = '2025-01-02'
  harga_satuan = 'Rp 85.000'
  diskon       = ''
  total        = 'Rp 170.000'

Tiga fakta terbaca dari keluaran tersebut. Pertama, pesanan.csv berisi 39 baris fisik tetapi 38 baris data — selisihnya adalah baris header. Kedua, satu baris events rusak dan kini punya alamat pasti: baris ke-116, terpotong di tengah nilai waktu. Ketiga, nilai mentah benar-benar mentah: harga_satuan berupa string 'Rp 85.000' dan diskon berupa string kosong, siap untuk dikoersi pada tahap berikutnya.

Periksa Ulang Jumlah Baris #

Angka 38 dan 116 pada keluaran di atas bukan sekadar informasi — keduanya menjadi angka patokan rekonsiliasi pada Bagian 7. Pemeriksaan test_baca_pesanan_baris_dan_kolom dan test_baca_events_jumlah pada tests/test_collect.py memastikan pembaca tidak pernah mengubah jumlah baris tanpa disadari, misalnya karena perubahan perilaku bawaan pandas.

Kedua sumber kini terbaca utuh dengan seluruh kotorannya. Pada Bagian 3 kita membersihkan kotoran tersebut satu per satu dengan aturan yang eksplisit.