Membaca Data dari Banyak Sumber

Sistem toko yang sesungguhnya tidak menyimpan data di satu tempat. Kasir cabang mengirim berkas CSV, aplikasi admin mengekspor JSON, laporan keuangan berada di Excel, sebagian pesanan lama tersimpan sebagai Parquet, dan data agregat duduk di database SQLite. Tahap extract pipeline yang baik harus sanggup menarik dari semua sumber itu, lalu menyatukannya menjadi satu DataFrame sebelum tahap transform dimulai. Chapter ini melatih pembacaan lima format sumber dengan reader pandas yang sesuai, semuanya dalam satu skrip yang bisa dijalankan berulang.

Peta Reader Pandas #

Setiap format punya fungsi pembaca dan kebiasaan dtype yang sedikit berbeda. Tabel berikut merangkum kelima sumber yang kita pakai.

Satu sumber boleh dijalankan dengan beberapa driver berbeda, dan pembacaan dari URL HTTP juga didukung oleh fungsi-fungsi ini; untuk contoh di sini kita memakai berkas lokal agar skrip tetap berjalan tanpa jaringan.

Lima Sumber dalam Satu Skrip #

Skrip berikut menulis data yang sama ke lima format, membacanya kembali dengan reader masing-masing, lalu menggabungkan hasilnya dengan pd.concat(). Seluruh berkas sumber dibuat di dalam TemporaryDirectory, sehingga disk otomatis bersih setelah skrip selesai.

"""Membaca pesanan dari CSV, JSON, Parquet, Excel, dan SQLite lalu menggabungkannya."""
import tempfile
from pathlib import Path

import pandas as pd
import sqlalchemy

PESANAN = pd.DataFrame({
    "id_pesanan": ["P001", "P002", "P003", "P004", "P005", "P006"],
    "cabang": ["Jakarta", "Bandung", "Surabaya", "Jakarta", "Bandung", "Surabaya"],
    "produk": ["Kopi Arabika", "Teh Melati", "Beras Premium",
               "Gula Pasir", "Kopi Arabika", "Minyak Goreng"],
    "jumlah": [2, 3, 1, 4, 2, 3],
    "harga": [45000, 18000, 62000, 14000, 45000, 32000],
})

KOLOM = ["id_pesanan", "cabang", "produk", "jumlah", "harga"]


def siapkan_sumber(root: Path) -> None:
    """Tulis data yang sama ke lima format sumber di folder root."""
    root.mkdir(parents=True, exist_ok=True)
    PESANAN.iloc[0:2].to_csv(root / "pesanan.csv", index=False)
    PESANAN.iloc[2:4].to_json(root / "pesanan.json", orient="records", lines=True)
    PESANAN.iloc[4:5].to_parquet(root / "pesanan.parquet", index=False)
    PESANAN.iloc[5:6].to_excel(root / "pesanan.xlsx", index=False)
    engine = sqlalchemy.create_engine(f"sqlite:///{root / 'toko.db'}")
    PESANAN.tail(1).to_sql("pesanan", engine, if_exists="replace", index=False)
    engine.dispose()


def baca_semua(root: Path) -> dict[str, pd.DataFrame]:
    """Baca kelima sumber dengan reader yang sesuai."""
    engine = sqlalchemy.create_engine(f"sqlite:///{root / 'toko.db'}")
    sumber = {
        "csv": pd.read_csv(root / "pesanan.csv"),
        "json": pd.read_json(root / "pesanan.json", lines=True),
        "parquet": pd.read_parquet(root / "pesanan.parquet"),
        "excel": pd.read_excel(root / "pesanan.xlsx"),
        "sql": pd.read_sql(f"SELECT {', '.join(KOLOM)} FROM pesanan", engine),
    }
    engine.dispose()
    return sumber


if __name__ == "__main__":
    with tempfile.TemporaryDirectory() as td:
        root = Path(td)
        siapkan_sumber(root)
        sumber = baca_semua(root)

        for nama, df in sumber.items():
            print(f"{nama:8s} baris={len(df)} dtype_harga={df['harga'].dtype}")

        gabungan = pd.concat(sumber.values(), ignore_index=True)
        print()
        print(f"gabungan: {gabungan.shape[0]} baris x {gabungan.shape[1]} kolom")
        print(gabungan.to_string(index=False))

Output yang Dihasilkan:

csv      baris=2 dtype_harga=int64
json     baris=2 dtype_harga=int64
parquet  baris=1 dtype_harga=int64
excel    baris=1 dtype_harga=int64
sql      baris=1 dtype_harga=int64

gabungan: 7 baris x 5 kolom
id_pesanan   cabang        produk  jumlah  harga
      P001  Jakarta  Kopi Arabika       2  45000
      P002  Bandung    Teh Melati       3  18000
      P003 Surabaya Beras Premium       1  62000
      P004  Jakarta    Gula Pasir       4  14000
      P005  Bandung  Kopi Arabika       2  45000
      P006 Surabaya Minyak Goreng       3  32000
      P006 Surabaya Minyak Goreng       3  32000

Yang perlu digarisbawahi dari skrip tersebut:

1. siapkan_sumber(root) memotong DataFrame induk dengan iloc dan menuliskan setiap potongan ke format berbeda. Pola ini meniru keadaan nyata, yaitu satu pesanan masuk lewat satu sumber tertentu.
2. baca_semua(root) memakai reader yang tepat per format. Untuk SQLite, fungsi pd.read_sql() menerima SQLAlchemy engine dari create_engine("sqlite:///...") dan menjalankan perintah SELECT yang kolomnya kita tulis eksplisit.
3. pd.concat(sumber.values(), ignore_index=True) menumpuk kelima DataFrame menjadi satu, dengan index baru berurutan dari nol.

Perhatikan baris P006 pada output: pesanan tersebut muncul dua kali karena sumber Excel dan SQLite memuat potongan yang sama. Duplikasi lintas sumber seperti ini adalah kejadian normal pada pipeline multi-sumber; tugas kita hanya memastikan tahap transform yang akan datang memiliki aturan pembersihan, dan itu dibahas pada chapter Transformasi ETL dengan Pandas.

Dtype yang Perlu Dikawal #

Hasil pd.concat() pada output menunjukkan dtype seragam: kolom teks menjadi str dan kolom angka menjadi int64 — keuntungan karena kelima sumber dibangun dari data yang sama. Pada data nyata, dua sumber bisa memberi dtype berbeda untuk kolom yang sama, misalnya jumlah terbaca int64 dari CSV tetapi float64 dari Excel karena ada sel kosong. Untuk itu, biasakan memeriksa dtype hasil gabungan:

• Tampilkan gabungan.dtypes sebelum lanjut ke tahap transform.
• Paksa tipe dengan astype() bila kolom kunci harus konsisten, misalnya sebelum merge().
• Periksa jumlah baris tiap sumber sebelum concat, sehingga sumber yang gagal terbaca cepat terlihat.

Setelah lima sumber menyatu menjadi satu DataFrame, tahap transform berikutnya bertugas merapikan isinya: menyeragamkan penulisan cabang, memetakan kode kategori, menghitung kolom total, dan membuang duplikat seperti baris P006 tadi.

Referensi resmi: pandas.read_csv , pandas.read_sql , user guide IO , serta dokumentasi SQLAlchemy untuk opsi engine database lain.