Dataset 19 baris hasil Bagian 3 kini terbukti sehat oleh empat gerbang validasi, tetapi data yang hanya berdiam di memori belum memberi manfaat apa pun kepada konsumennya. Tahap terakhir pipeline ETL adalah load: menyalurkan hasil yang sudah bersih ke format yang dipakai pihak di luar pipeline. Pada bagian penutup proyek ini kita mengekspor satu dataset ke tiga tujuan berbeda — Parquet untuk dataset lengkap, CSV untuk laporan ringkasan omzet per kota, dan SQLite untuk aplikasi lain yang membaca tabel SQL. Sebagai penutup jejak audit, skrip juga mencetak rekap aliran baris dari setiap sumber sehingga angka 25 masuk dan 19 keluar yang diperkenalkan di Bagian 2 akhirnya terurai per sumber.

Satu Dataset untuk Tiga Konsumen #

Pemilihan format ekspor mengikuti kebutuhan konsumennya, bukan preferensi penulis pipeline. Format kolom Parquet menyimpan dtype dalam metadata dan efisien untuk dataset besar; CSV dibaca manusia dan aplikasi spreadsheet; tabel SQLite di-query aplikasi lain tanpa library pandas.

Tujuan eksporFormatIsiKonsumen yang dilayani
hasil_bersih.parquetParquet (kolom)dataset lengkap 19 baris x 9 kolompipeline dan analisis berikutnya
ringkasan_omzet.csvCSVomzet per kota hasil groupby()tim keuangan, aplikasi spreadsheet
tabel hasil_bersih di hasil.dbSQLitedataset lengkap via to_sql()aplikasi yang membaca SQL

Agregasi omzet per kota dibangun dengan groupby("kota", as_index=False).agg(...): kolom transaksi menghitung jumlah baris dan kolom omzet menjumlahkan total, lalu sort_values("omzet", ascending=False) mengurutkan kota dari kontributor terbesar. Hasilnya Denpasar memimpin dengan omzet 1010500 dari 7 transaksi, disusul Bandung 698000 dan Jakarta 298000.

Skrip Ekspor Lengkap #

Skrip ekspor_hasil.py menuntaskan proyek: fungsi transformasi() yang identik dengan Bagian 3 menghasilkan DataFrame bersih, fungsi rekap_pipeline() menghitung aliran baris per sumber, lalu tiga ekspor dijalankan dan sekaligus dibaca ulang sebagai bukti.

python 
"""Ekspor hasil pipeline ETL ke Parquet, CSV ringkasan, dan tabel SQLite."""
from __future__ import annotations

import tempfile
from pathlib import Path

import pandas as pd
from sqlalchemy import create_engine

from data_sumber import muat_csv, muat_json, muat_sqlite


def transformasi(
    csv_raw: pd.DataFrame, json_raw: pd.DataFrame, sqlite_raw: pd.DataFrame
) -> pd.DataFrame:
    """Seluruh tahap transformasi Bagian 2, tanpa pencetakan."""
    csv_df = csv_raw.rename(columns={"kode_trx": "id_transaksi"})
    csv_df["tanggal"] = pd.to_datetime(csv_df["tanggal"], format="%d/%m/%Y")
    csv_df["kota"] = csv_df["kota"].str.strip().str.title()
    csv_df = csv_df[csv_df["harga_satuan"] > 0]
    csv_df["sumber"] = "laporan_csv"

    json_df = json_raw.rename(
        columns={
            "order_id": "id_transaksi",
            "item": "produk",
            "kota_tujuan": "kota",
            "jml_barang": "qty",
        }
    )
    json_df = json_df[json_df["qty"] > 0]
    json_df["harga_satuan"] = json_df["harga_total"] // json_df["qty"]
    json_df["tanggal"] = pd.to_datetime(json_df["created_at"]).dt.normalize()
    json_df["kota"] = json_df["kota"].str.strip().str.title()
    json_df = json_df.drop(columns=["created_at", "harga_total"])
    json_df["sumber"] = "marketplace_json"

    sqlite_df = sqlite_raw.rename(
        columns={
            "trx_id": "id_transaksi",
            "tgl_trx": "tanggal",
            "nama_barang": "produk",
            "wilayah": "kota",
            "jml": "qty",
            "harga": "harga_satuan",
        }
    )
    sqlite_df["tanggal"] = pd.to_datetime(sqlite_df["tanggal"])
    sqlite_df = sqlite_df.drop(columns="kasir")
    sqlite_df["sumber"] = "kasir_sqlite"

    kolom_inti = [
        "id_transaksi", "tanggal", "produk", "kategori", "kota", "qty", "harga_satuan",
    ]
    gabung = pd.concat(
        [
            csv_df[kolom_inti + ["sumber"]],
            json_df[kolom_inti + ["sumber"]],
            sqlite_df[kolom_inti + ["sumber"]],
        ],
        ignore_index=True,
    )
    bersih = gabung.drop_duplicates(subset="id_transaksi", keep="last")
    bersih = bersih.reset_index(drop=True)
    bersih["total"] = bersih["qty"] * bersih["harga_satuan"]
    return bersih[
        ["id_transaksi", "tanggal", "produk", "kategori", "kota",
         "qty", "harga_satuan", "total", "sumber"]
    ]


def rekap_pipeline(
    csv_raw: pd.DataFrame,
    json_raw: pd.DataFrame,
    sqlite_raw: pd.DataFrame,
    hasil: pd.DataFrame,
) -> pd.DataFrame:
    """Tabel aliran baris: masuk, dibuang, dan keluar per sumber."""
    baris = []
    for nama, raw in (
        ("laporan_csv", csv_raw),
        ("marketplace_json", json_raw),
        ("kasir_sqlite", sqlite_raw),
    ):
        keluar = int((hasil["sumber"] == nama).sum())
        baris.append(
            {
                "sumber": nama,
                "masuk": len(raw),
                "dibuang": len(raw) - keluar,
                "keluar": keluar,
            }
        )
    return pd.DataFrame(baris)


def main() -> None:
    with tempfile.TemporaryDirectory() as tmp:
        folder = Path(tmp)
        csv_raw = muat_csv(folder)
        json_raw = muat_json()
        sqlite_raw = muat_sqlite(folder)
        hasil = transformasi(csv_raw, json_raw, sqlite_raw)

        rekap = rekap_pipeline(csv_raw, json_raw, sqlite_raw, hasil)
        print("== Ringkasan pipeline ==")
        print(rekap.to_string(index=False))
        print(
            f"total: {rekap['masuk'].sum()} masuk,"
            f" {rekap['dibuang'].sum()} dibuang, {rekap['keluar'].sum()} keluar"
        )

        omzet_kota = (
            hasil.groupby("kota", as_index=False)
            .agg(transaksi=("id_transaksi", "count"), omzet=("total", "sum"))
            .sort_values("omzet", ascending=False)
            .reset_index(drop=True)
        )
        print("\n== Omzet per kota ==")
        print(omzet_kota.to_string(index=False))

        # Ekspor 1: Parquet untuk dataset lengkap.
        path_parquet = folder / "hasil_bersih.parquet"
        hasil.to_parquet(path_parquet)
        baca_parquet = pd.read_parquet(path_parquet)
        print("\n== Ekspor dan pemeriksaan baca ulang ==")
        print(
            f"hasil_bersih.parquet : {baca_parquet.shape[0]} baris x"
            f" {baca_parquet.shape[1]} kolom (baca ulang OK)"
        )

        # Ekspor 2: CSV ringkasan untuk laporan.
        path_csv = folder / "ringkasan_omzet.csv"
        omzet_kota.to_csv(path_csv, index=False)
        baca_csv = pd.read_csv(path_csv)
        print(f"ringkasan_omzet.csv  : {len(baca_csv)} baris omzet per kota (baca ulang OK)")

        # Ekspor 3: tabel SQLite untuk aplikasi lain.
        engine = create_engine(f"sqlite:///{folder / 'hasil.db'}")
        hasil.to_sql("hasil_bersih", engine, if_exists="replace", index=False)
        hitung = pd.read_sql("SELECT COUNT(*) AS jumlah FROM hasil_bersih", engine)
        engine.dispose()
        print(f"hasil.db hasil_bersih: {int(hitung['jumlah'].iloc[0])} baris (if_exists=\"replace\")")

    print("\nBerkas sementara (.parquet, .csv, .db) terhapus bersih folder sementaranya.")


if __name__ == "__main__":
    main()

Output yang Dihasilkan:

text 
== Ringkasan pipeline ==
          sumber  masuk  dibuang  keluar
     laporan_csv     10        5       5
marketplace_json      8        1       7
    kasir_sqlite      7        0       7
total: 25 masuk, 6 dibuang, 19 keluar

== Omzet per kota ==
    kota  transaksi   omzet
Denpasar          7 1010500
 Bandung          5  698000
 Jakarta          3  298000
   Depok          1  115500
  Bekasi          2   76500
   Bogor          1   62500

== Ekspor dan pemeriksaan baca ulang ==
hasil_bersih.parquet : 19 baris x 9 kolom (baca ulang OK)
ringkasan_omzet.csv  : 6 baris omzet per kota (baca ulang OK)
hasil.db hasil_bersih: 19 baris (if_exists="replace")

Berkas sementara (.parquet, .csv, .db) terhapus bersih folder sementaranya.

Mari kita telaah bagian-bagian pentingnya:

  1. rekap_pipeline(): tabel rekap membandingkan len(raw) setiap sumber dengan jumlah baris hasil yang bertanda sumber sama. Lima baris laporan_csv dibuang (satu harga 0, tiga duplikat lintas sumber, satu salinan ganda), satu baris marketplace_json dibuang karena qty 0, dan kasir_sqlite lepas tanpa kehilangan karena keep="last" mempertahankan versi resminya.
  2. hasil.to_parquet(path_parquet): to_parquet() menulis DataFrame sebagai berkas kolom via pyarrow. Pemeriksaan baca ulang dengan read_parquet() mencetak 19 baris x 9 kolom — dtype str, datetime64[us], dan int64 tersimpan persis di metadata berkas.
  3. omzet_kota.to_csv(path_csv, index=False): argumen index=False mencegah nomor index ikut tersimpan sebagai kolom tambahan tanpa nama. Baca ulang dengan read_csv() menghasilkan 6 baris omzet per kota.
  4. hasil.to_sql("hasil_bersih", engine, if_exists="replace", index=False): to_sql() menulis seluruh DataFrame ke tabel SQLite melalui engine SQLAlchemy, lalu query SELECT COUNT(*) membaca kembali 19 baris dari hasil_bersih.

Empat bagian proyek kini tertutup lengkap: Bagian 1 mengekstrak tiga sumber berformat campur, Bagian 2 menyeragamkan skema dan menggabungkan, Bagian 3 memvalidasi hasil, dan Bagian 4 mengekspor data bersih. Dataset Toko Nusantara Mart bergerak dari 25 baris mentah menjadi tiga berkas siap pakai dengan rekap aliran baris yang dapat dijalankan ulang kapan pun. Pola yang sama — standarisasi skema, deduplikasi lintas sumber, gerbang validasi, dan ekspor multi-tujuan — berlaku untuk pipeline ETL mana pun yang akan Anda bangun berikutnya.

Course ini ditutup oleh proyek terakhir, Analisis Deret Waktu Penjualan, yang mengolah data penjualan bersih dengan kolom tanggal sebagai sumbu waktunya. Kita memulainya pada Bagian 1: Mempersiapkan Data Waktu.

Sumber bacaan resmi: pandas.DataFrame.to_parquet , pandas.DataFrame.to_sql , dan pandas.DataFrame.to_csv .