Ekspor Data Hasil Transformasi

Pipeline yang lolos validasi pada chapter Validasi Data Pipeline kini menyimpan enam baris pesanan yang tepercaya, tetapi data tersebut masih tinggal di memori. Tim lain tidak membaca memori: tim keuangan membuka Excel, dashboard membaca database, dan aplikasi downstream menuntut CSV atau JSON. Tahap load bertugas menuliskan satu hasil transformasi ke sejumlah format tujuan sekaligus, dengan nama berkas yang konsisten agar penerima dapat mengotomatiskan sisi mereka. Chapter ini mengekspor hasil pipeline Toko Berkah Abadi ke lima format, memverifikasi hasil dengan membaca ulang, dan membahas perilaku if_exists yang sering mengejutkan pemula pada to_sql().

Memilih Format Tujuan #

Setiap format punya peran yang berbeda pada rantai distribusi data:

Tujuan	Method penulisan	Cocok untuk
CSV	`to_csv()`	Pertukaran universal, dibuka di mana saja
Parquet	`to_parquet()`	Arsip kolom biner, dtype dan kompresi terjaga
Excel	`to_excel()`	Pemakai akhir yang bekerja dengan spreadsheet
JSON	`to_json()`	API dan aplikasi yang membaca data terstruktur
SQLite	`to_sql()`	Dashboard dan query lanjutan lewat SQL

Semua method di atas menerima path berkas, dan to_sql() menerima SQLAlchemy engine alih-alih path. Penamaan berkas mengikuti satu pola: pesanan_bersih plus ekstensi format — nama dasar yang sama membuat lima keluaran mudah dikenali sebagai satu produk pipeline.

Skrip Ekspor dan Verifikasi Baca Ulang #

Mengekspor tanpa memeriksa hasil sama dengan mengirim paket tanpa konfirmasi penerimaan. Script berikut menulis kelima format ke dalam TemporaryDirectory, lalu membaca setiap hasil kembali dan menghitung jumlah barisnya.

python

"""Ekspor hasil transformasi ke lima format dengan penamaan konsisten."""
import tempfile
from pathlib import Path

import pandas as pd
import sqlalchemy


def hasil_transformasi() -> pd.DataFrame:
    """Data bersih yang sudah lolos validasi."""
    return 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"],
        "kategori": ["Minuman", "Minuman", "Bahan Pokok",
                     "Bahan Pokok", "Minuman", "Bahan Pokok"],
        "jumlah": [2, 3, 1, 4, 2, 3],
        "harga_satuan": [45000, 18000, 62000, 14000, 45000, 32000],
        "total": [90000, 54000, 62000, 56000, 90000, 96000],
    })


def ekspor_semua(hasil: pd.DataFrame, root: Path) -> None:
    """Tulis satu hasil ke beberapa format dengan nama berkas seragam."""
    root.mkdir(parents=True, exist_ok=True)
    hasil.to_csv(root / "pesanan_bersih.csv", index=False)
    hasil.to_parquet(root / "pesanan_bersih.parquet", index=False)
    hasil.to_excel(root / "pesanan_bersih.xlsx", sheet_name="pesanan", index=False)
    hasil.to_json(root / "pesanan_bersih.json", orient="records", indent=2)
    engine = sqlalchemy.create_engine(f"sqlite:///{root / 'toko.db'}")
    hasil.to_sql("pesanan", engine, if_exists="replace", index=False)
    engine.dispose()


def verifikasi(root: Path) -> dict[str, int]:
    """Baca kembali setiap hasil dan hitung jumlah barisnya."""
    engine = sqlalchemy.create_engine(f"sqlite:///{root / 'toko.db'}")
    try:
        return {
            "csv": len(pd.read_csv(root / "pesanan_bersih.csv")),
            "parquet": len(pd.read_parquet(root / "pesanan_bersih.parquet")),
            "excel": len(pd.read_excel(root / "pesanan_bersih.xlsx")),
            "json": len(pd.read_json(root / "pesanan_bersih.json")),
            "sqlite": len(pd.read_sql("SELECT * FROM pesanan", engine)),
        }
    finally:
        engine.dispose()


if __name__ == "__main__":
    hasil = hasil_transformasi()
    with tempfile.TemporaryDirectory() as td:
        root = Path(td) / "hasil"
        ekspor_semua(hasil, root)
        print("== Ekspor dan baca ulang ==")
        for format_ekspor, baris in verifikasi(root).items():
            print(f"{format_ekspor:7s}: {baris} baris")
        print()
        print("== Dua baris pertama pesanan_bersih.csv ==")
        for baris in (root / "pesanan_bersih.csv").read_text(
            encoding="utf-8"
        ).splitlines()[:2]:
            print(baris)
        print()

        engine = sqlalchemy.create_engine(f"sqlite:///{root / 'toko.db'}")
        try:
            print("== Perilaku if_exists pada to_sql ==")
            hasil.to_sql("pesanan", engine, if_exists="replace", index=False)
            print(f"setelah replace : {len(pd.read_sql('SELECT * FROM pesanan', engine))}")
            hasil.to_sql("pesanan", engine, if_exists="append", index=False)
            print(f"setelah append  : {len(pd.read_sql('SELECT * FROM pesanan', engine))}")
            hasil.to_sql("pesanan", engine, if_exists="replace", index=False)
            print(f"replace lagi    : {len(pd.read_sql('SELECT * FROM pesanan', engine))}")
        finally:
            engine.dispose()

Output yang Dihasilkan:

text
== Ekspor dan baca ulang ==
csv    : 6 baris
parquet: 6 baris
excel  : 6 baris
json   : 6 baris
sqlite : 6 baris

== Dua baris pertama pesanan_bersih.csv ==
id_pesanan,cabang,produk,kategori,jumlah,harga_satuan,total
P001,Jakarta,Kopi Arabika,Minuman,2,45000,90000

== Perilaku if_exists pada to_sql ==
setelah replace : 6
setelah append  : 12
replace lagi    : 6

Keputusan teknis yang perlu dipahami:

to_csv(..., index=False) tidak menulis index. Bila argumen ini lupa, CSV akan memuat satu kolom tanpa nama berisi angka 0 sampai 5 — kolom sisa yang mengganggu pembaca berikutnya.
to_excel(..., sheet_name="pesanan") menamai sheet secara eksplisit; tanpa argumen ini, pandas memakai nama bawaan Sheet1.
to_json(orient="records", indent=2) menulis daftar objek JSON, satu objek per baris data, dengan indentasi yang memudahkan pemeriksaan manual.
to_sql("pesanan", engine, if_exists="replace", index=False) menulis tabel pesanan ke SQLite. Argumen if_exists="replace" menjatuhkan tabel lama lalu membuat yang baru.
verifikasi() membaca ulang kelima hasil dan mengembalikan jumlah baris; kelima angka 6 pada output memastikan tidak ada hasil yang terpotong.

Baca ulang ini sengaja diletakkan di dalam skrip, bukan sekadar dijalankan manual sekali. Verifikasi otomatis berarti kelima format diperiksa setiap kali pipeline berjalan, bukan hanya saat pipeline pertama ditulis.

Perilaku if_exists pada to_sql #

Output bagian terakhir memperlihatkan tiga eksekusi to_sql() berurutan ke tabel yang sama:

if_exists="replace": tabel pesanan dibuat ulang, hasilnya 6 baris.
if_exists="append": enam baris baru ditambahkan di atas enam baris lama, hasilnya 12 baris.
replace lagi: tabel dibuat ulang, angka kembali ke 6.

Pola replace → append → replace menunjukkan secara langsung bahwa append menggandakan data bila pipeline berjalan dua kali tanpa pengelolaan. Pipeline yang dijadwalkan berjalan harian sebaiknya memakai replace untuk tabel penuh, atau append hanya bila idempotensi dijauhkan dengan desain lain — misalnya tabel staging yang dibersihkan sebelum append.

Satu hasil transformasi kini sampai ke lima tujuan dengan benar dan terverifikasi. Langkah terakhir untuk membuat pipeline benar-benar produktif adalah membungkus seluruh tahapan menjadi satu skrip yang dijalankan dari terminal — lengkap dengan opsi baris perintah — sehingga jadwal otomatis cukup memanggilnya tanpa menyentuh kode.

Referensi resmi: pandas.DataFrame.to_sql , pandas.DataFrame.to_excel , dan pandas.DataFrame.to_csv .