Toko Nusantara Mart menjual produk yang sama melalui tiga jalur: Cabang A di Bandung dengan laporan penjualan CSV, Cabang B di Denpasar yang transaksinya tersimpan di tabel SQLite milik sistem kasir, dan sebuah kanal marketplace yang mencatat pesanan dalam format JSON. Tim keuangan meminta satu tabel gabungan berisi seluruh transaksi Januari hingga Maret 2026, tetapi ketiga sistem menamai kolom dengan gaya berbeda, menulis tanggal dengan format berbeda, dan menyimpan harga dengan makna berbeda. Menggabungkan sumber seperti inilah pekerjaan inti pipeline ETL (extract, transform, load), dan pandas menyediakan semua peralatannya. Pada bagian pertama proyek ini kita mengekstrak ketiga sumber memakai read_csv(), read_json(), dan read_sql(), memeriksa strukturnya, lalu menyusun daftar perbedaan format sebagai kontrak kerja Bagian 2.

Tiga Sumber dengan Tiga Format #

Ketiga sumber didefinisikan pada modul data_sumber.py yang menyertai bagian ini. Isinya dibangun dari generator deterministik dengan seed 88, sehingga angkanya identik di komputer mana pun. Cerita datanya sengaja dirancang agar proses ETL bermakna:

  • Laporan CSV adalah ekspor gabungan: enam transaksi Cabang A, plus tiga transaksi Cabang B yang sudah tersinkron ke laporan regional. Salinan ganda satu baris juga ikut terbawa di dalam laporan itu sendiri.
  • JSON marketplace menyimpan harga_total per pesanan, bukan harga_satuan per barang, dan label kota ditulis tanpa standar: jakarta, BEKASI, tangerang.
  • Tabel SQLite Cabang B memakai nama kolom sendiri (trx_id, tgl_trx, jml) dan memiliki kolom ekstra kasir yang tidak dibutuhkan skema akhir.
SumberSistem asalFormatKeanehan yang ditemukan
laporan_penjualan.csvlaporan gabungan cabangCSV, tanggal DD/MM/YYYYbaris ganda, satu harga_satuan 0
JSON pesanan marketplacekanal onlineJSON records, created_at ISO + jamharga_total, label kota campur, satu jml_barang 0
tabel transaksi di toko_b.dbkasir Cabang BSQLite, tanggal teks YYYY-MM-DDkolom ekstra kasir, nama kolom berbeda

Modul Definisi Sumber #

Modul data_sumber.py memuat konstanta CSV_TEKS, JSON_TEKS, dan DATA_SQLITE beserta tiga fungsi pemuat. Fungsi muat_csv() menulis laporan ke sebuah folder lalu membacanya dengan read_csv(); fungsi muat_json() membaca teks JSON melalui StringIO; fungsi muat_sqlite() membuat berkas toko_b.db dengan modul sqlite3 untuk mensimulasikan database kasir yang sudah berisi data, lalu membaca tabelnya dengan read_sql() melalui engine SQLAlchemy. Menjalankan modul ini langsung hanya mencetak ringkasan singkat; pemuatan sesungguhnya dilakukan skrip pada bagian berikutnya.

python 
"""Definisi tiga sumber data proyek ETL Toko Nusantara Mart.

Tiga sistem penjualan menyimpan transaksi yang perlu digabung:
  1. laporan CSV gabungan dua cabang (kolom kode_trx, tanggal DD/MM/YYYY, dst.),
  2. JSON pesanan marketplace (kolom order_id, created_at, harga_total),
  3. tabel SQLite sistem kasir Cabang B (kolom trx_id, tgl_trx, ..., kasir).
Isi ketiga sumber dibangun secara deterministik dari seed 88 sehingga hasilnya
identik di komputer mana pun.
"""
from __future__ import annotations

import json
import sqlite3
from io import StringIO
from pathlib import Path

import numpy as np
import pandas as pd
from sqlalchemy import create_engine

SEED = 88

HARGA_SATUAN = {
    "Kopi Robusta Bubuk 200g": 38500,
    "Teh Celup Melati 25": 12500,
    "Gula Pasir 1kg": 16000,
    "Beras Pandan Wangi 5kg": 72000,
    "Minyak Goreng 2L": 36500,
    "Mie Goreng Instan 85g": 3500,
}

KATEGORI = {
    "Kopi Robusta Bubuk 200g": "Minuman",
    "Teh Celup Melati 25": "Minuman",
    "Gula Pasir 1kg": "Bahan Pokok",
    "Beras Pandan Wangi 5kg": "Bahan Pokok",
    "Minyak Goreng 2L": "Bahan Pokok",
    "Mie Goreng Instan 85g": "Bahan Pokok",
}


def _bangun_sumber() -> tuple[str, str, list[tuple[str, ...]]]:
    """Bangun isi ketiga sumber secara deterministik dari seed 88."""
    rng = np.random.default_rng(SEED)

    def qty() -> int:
        return int(rng.integers(1, 7))

    # Sumber 1: laporan CSV gabungan. Tiga transaksi Cabang B yang sudah
    # tersinkron ikut terbawa di sini (nanti terbaca duplikat lintas sumber).
    baris_csv = [
        ("TRX-A-001", "14/01/2026", "Kopi Robusta Bubuk 200g", "Bandung"),
        ("TRX-A-002", "16/01/2026", "Gula Pasir 1kg", "Bandung"),
        ("TRX-A-003", "21/01/2026", "Beras Pandan Wangi 5kg", "Bandung"),
        ("TRX-A-004", "25/01/2026", "Minyak Goreng 2L", "Bandung"),
        ("TRX-A-005", "18/02/2026", "Teh Celup Melati 25", "Bandung"),
        ("TRX-A-006", "21/02/2026", "Mie Goreng Instan 85g", "Bandung"),
        ("TRX-B-001", "15/01/2026", "Kopi Robusta Bubuk 200g", "Denpasar"),
        ("TRX-B-002", "20/01/2026", "Beras Pandan Wangi 5kg", "Denpasar"),
        ("TRX-B-003", "17/02/2026", "Minyak Goreng 2L", "Denpasar"),
    ]
    jalur_csv = ["kode_trx,tanggal,produk,kategori,kota,qty,harga_satuan"]
    qty_cabang_b: dict[str, int] = {}
    for trx_id, tanggal, produk, kota in baris_csv:
        harga = HARGA_SATUAN[produk]
        if trx_id == "TRX-A-004":
            harga = 0  # kesalahan perekaman: harga satuan tercatat nol
        jml = qty()
        if trx_id.startswith("TRX-B-"):
            qty_cabang_b[trx_id] = jml  # dipakai ulang oleh sumber SQLite
        jalur_csv.append(
            f"{trx_id},{tanggal},{produk},{KATEGORI[produk]},{kota},{jml},{harga}"
        )
    salinan = next(b for b in jalur_csv if b.startswith("TRX-B-002"))
    jalur_csv.append(salinan)  # baris ganda di dalam laporan itu sendiri
    csv_teks = "\n".join(jalur_csv)

    # Sumber 2: JSON pesanan marketplace dengan format berbeda.
    baris_json = [
        ("WEB-001", "2026-01-03T09:15:00", "Kopi Robusta Bubuk 200g", "jakarta"),
        ("WEB-002", "2026-01-05T14:02:00", "Gula Pasir 1kg", "Jakarta"),
        ("WEB-003", "2026-01-09T10:41:00", "Mie Goreng Instan 85g", "BEKASI"),
        ("WEB-004", "2026-02-02T19:30:00", "Teh Celup Melati 25", "Bogor"),
        ("WEB-005", "2026-02-05T11:05:00", "Beras Pandan Wangi 5kg", "tangerang"),
        ("WEB-006", "2026-02-14T16:20:00", "Gula Pasir 1kg", "Jakarta"),
        ("WEB-007", "2026-03-04T08:55:00", "Minyak Goreng 2L", "Bekasi"),
        ("WEB-008", "2026-03-11T13:47:00", "Kopi Robusta Bubuk 200g", "Depok"),
    ]
    catatan_json = []
    for order_id, created_at, produk, kota in baris_json:
        jml = qty()
        if order_id == "WEB-005":
            jml = 0  # pesanan gagal dibayar: jumlah barang nol
        catatan_json.append(
            {
                "order_id": order_id,
                "created_at": created_at,
                "item": produk,
                "kategori": KATEGORI[produk],
                "kota_tujuan": kota,
                "jml_barang": jml,
                "harga_total": jml * HARGA_SATUAN[produk],
            }
        )
    json_teks = json.dumps(catatan_json, indent=2)

    # Sumber 3: tabel SQLite milik sistem kasir Cabang B (kolom ekstra kasir).
    baris_b = [
        ("TRX-B-001", "2026-01-15", "Kopi Robusta Bubuk 200g", "Rina"),
        ("TRX-B-002", "2026-01-20", "Beras Pandan Wangi 5kg", "Budi"),
        ("TRX-B-003", "2026-02-17", "Minyak Goreng 2L", "Rina"),
        ("TRX-B-004", "2026-02-19", "Teh Celup Melati 25", "Siti"),
        ("TRX-B-005", "2026-02-23", "Gula Pasir 1kg", "Budi"),
        ("TRX-B-006", "2026-03-06", "Mie Goreng Instan 85g", "Siti"),
        ("TRX-B-007", "2026-03-15", "Kopi Robusta Bubuk 200g", "Rina"),
    ]
    data_sqlite = [
        (
            trx_id,
            tgl,
            produk,
            KATEGORI[produk],
            "Denpasar",
            qty_cabang_b.get(trx_id, qty()),
            HARGA_SATUAN[produk],
            kasir,
        )
        for trx_id, tgl, produk, kasir in baris_b
    ]
    return csv_teks, catatan_json, data_sqlite


CSV_TEKS, PESANAN_JSON, DATA_SQLITE = _bangun_sumber()
JSON_TEKS = json.dumps(PESANAN_JSON, indent=2)


def muat_csv(folder: Path) -> pd.DataFrame:
    """Tulis laporan CSV ke folder lalu baca dengan read_csv."""
    path = folder / "laporan_penjualan.csv"
    path.write_text(CSV_TEKS, encoding="utf-8")
    return pd.read_csv(path)


def muat_json() -> pd.DataFrame:
    """Baca JSON pesanan marketplace dari teks melalui StringIO."""
    return pd.read_json(StringIO(JSON_TEKS), convert_dates=False)


def muat_sqlite(folder: Path) -> pd.DataFrame:
    """Buat database kasir Cabang B lalu baca tabelnya dengan read_sql."""
    path = folder / "toko_b.db"
    koneksi = sqlite3.connect(path)
    try:
        koneksi.execute(
            "CREATE TABLE transaksi (trx_id TEXT, tgl_trx TEXT, nama_barang TEXT,"
            " kategori TEXT, wilayah TEXT, jml INTEGER, harga INTEGER, kasir TEXT)"
        )
        koneksi.executemany(
            "INSERT INTO transaksi VALUES (?, ?, ?, ?, ?, ?, ?, ?)", DATA_SQLITE
        )
        koneksi.commit()
    finally:
        koneksi.close()
    engine = create_engine(f"sqlite:///{path}")
    try:
        return pd.read_sql("SELECT * FROM transaksi", engine)
    finally:
        engine.dispose()


if __name__ == "__main__":
    print(f"laporan_csv      : {len(CSV_TEKS.splitlines()) - 1} baris")
    print(f"marketplace_json : {len(PESANAN_JSON)} pesanan")
    print(f"kasir_sqlite     : {len(DATA_SQLITE)} transaksi")
    print(f"produk referensi : {len(HARGA_SATUAN)} barang dengan harga tetap")

Output yang Dihasilkan:

text 
laporan_csv      : 10 baris
marketplace_json : 8 pesanan
kasir_sqlite     : 7 transaksi
produk referensi : 6 barang dengan harga tetap

Memuat Ketiga Sumber #

Skrip muat_sumber.py memanggil ketiga pemuat, mencetak dimensi, dtypes, contoh baris tiap sumber, lalu menutup dengan rekap perbedaan format.

python 
"""Memuat tiga sumber transaksi Toko Nusantara Mart dan memetakan perbedaannya."""
from __future__ import annotations

import tempfile
from pathlib import Path

from data_sumber import muat_csv, muat_json, muat_sqlite


def main() -> None:
    with tempfile.TemporaryDirectory() as tmp:
        folder = Path(tmp)
        csv_df = muat_csv(folder)
        json_df = muat_json()
        sqlite_df = muat_sqlite(folder)

    for nama, df in (
        ("laporan_csv", csv_df),
        ("marketplace_json", json_df),
        ("kasir_sqlite", sqlite_df),
    ):
        print(f"== {nama}: {df.shape[0]} baris x {df.shape[1]} kolom ==")
        for kolom, tipe in df.dtypes.items():
            print(f"  {kolom:12s} {tipe}")
        print(df.head(3).to_string(index=False))
        print()

    irisan = set(csv_df["kode_trx"]) & set(sqlite_df["trx_id"])
    print("== Perbedaan format yang harus diseragamkan ==")
    print("nama kolom id   : kode_trx (csv), order_id (json), trx_id (sqlite)")
    print("format tanggal  : DD/MM/YYYY (csv), ISO + jam (json), YYYY-MM-DD teks (sqlite)")
    print("makna harga     : harga_satuan (csv, sqlite) vs harga_total (json)")
    print(f"penulisan kota  : {json_df['kota_tujuan'].nunique()} bentuk penulisan di JSON")
    print("kolom ekstra    : kasir di SQLite tidak dibutuhkan skema akhir")
    print(
        f"baris tidak valid: {(csv_df['harga_satuan'] == 0).sum()} harga 0 di CSV,"
        f" {(json_df['jml_barang'] == 0).sum()} qty 0 di JSON"
    )
    print(f"irisan sumber   : {len(irisan)} transaksi Cabang B ada di CSV dan SQLite sekaligus")


if __name__ == "__main__":
    main()

Output yang Dihasilkan:

text 
== laporan_csv: 10 baris x 7 kolom ==
  kode_trx     str
  tanggal      str
  produk       str
  kategori     str
  kota         str
  qty          int64
  harga_satuan int64
 kode_trx    tanggal                  produk    kategori    kota  qty  harga_satuan
TRX-A-001 14/01/2026 Kopi Robusta Bubuk 200g     Minuman Bandung    4         38500
TRX-A-002 16/01/2026          Gula Pasir 1kg Bahan Pokok Bandung    2         16000
TRX-A-003 21/01/2026  Beras Pandan Wangi 5kg Bahan Pokok Bandung    6         72000

== marketplace_json: 8 baris x 7 kolom ==
  order_id     str
  created_at   str
  item         str
  kategori     str
  kota_tujuan  str
  jml_barang   int64
  harga_total  int64
order_id          created_at                    item    kategori kota_tujuan  jml_barang  harga_total
 WEB-001 2026-01-03T09:15:00 Kopi Robusta Bubuk 200g     Minuman     jakarta           4       154000
 WEB-002 2026-01-05T14:02:00          Gula Pasir 1kg Bahan Pokok     Jakarta           4        64000
 WEB-003 2026-01-09T10:41:00   Mie Goreng Instan 85g Bahan Pokok      BEKASI           1         3500

== kasir_sqlite: 7 baris x 8 kolom ==
  trx_id       str
  tgl_trx      str
  nama_barang  str
  kategori     str
  wilayah      str
  jml          int64
  harga        int64
  kasir        str
   trx_id    tgl_trx             nama_barang    kategori  wilayah  jml  harga kasir
TRX-B-001 2026-01-15 Kopi Robusta Bubuk 200g     Minuman Denpasar    6  38500  Rina
TRX-B-002 2026-01-20  Beras Pandan Wangi 5kg Bahan Pokok Denpasar    5  72000  Budi
TRX-B-003 2026-02-17        Minyak Goreng 2L Bahan Pokok Denpasar    3  36500  Rina

== Perbedaan format yang harus diseragamkan ==
nama kolom id   : kode_trx (csv), order_id (json), trx_id (sqlite)
format tanggal  : DD/MM/YYYY (csv), ISO + jam (json), YYYY-MM-DD teks (sqlite)
makna harga     : harga_satuan (csv, sqlite) vs harga_total (json)
penulisan kota  : 7 bentuk penulisan di JSON
kolom ekstra    : kasir di SQLite tidak dibutuhkan skema akhir
baris tidak valid: 1 harga 0 di CSV, 1 qty 0 di JSON
irisan sumber   : 3 transaksi Cabang B ada di CSV dan SQLite sekaligus

Mari kita telaah bagian-bagian pentingnya:

  1. with tempfile.TemporaryDirectory() as tmp: berkas CSV dan database toko_b.db dibuat di folder sementara yang terhapus sendiri ketika blok with berakhir, sehingga proyek tidak meninggalkan artefak. Pada pekerjaan nyata, ganti dengan folder sumber Anda.
  2. Tiga pemanggilan pemuat: setiap sumber kembali sebagai DataFrame mentah. Kolom teks bertipe str — perilaku bawaan pandas 3.0, bukan lagi object — sementara kolom angka bertipe int64.
  3. Perulangan pencetakan: df.shape dan df.dtypes memperlihatkan bahwa ketiga sumber punya jumlah dan nama kolom yang berbeda, dan df.head(3) menampilkan contoh format tanggal serta label kota yang tidak seragam.
  4. Irisan set(csv_df["kode_trx"]) & set(sqlite_df["trx_id"]): tiga id transaksi Cabang B muncul di dua sumber sekaligus. Irisan inilah duplikat lintas sumber yang akan dibuang di Bagian 2.

Kontrak Kerja Menuju Bagian 2 #

Rekap di akhir keluaran adalah kontrak kerja Bagian 2: ganti nama kolom ketiga sumber ke satu skema bersama, ubah tiga format tanggal menjadi datetime64[us], turunkan harga_satuan dari harga_total di sumber JSON, seragamkan penulisan kota, buang kolom kasir, dan lewati baris yang tidak valid. Pola membaca banyak sumber sudah diperkenalkan pada chapter Membaca Data dari Banyak Sumber di section Praktik Data Engineering; di sini kita menerapkannya pada skema yang berbeda-beda dan menambahkan dimensi baru: duplikasi antar sistem. Setelah daftar perbedaan ini jelas, Bagian 2 menyatukan ketiga sumber menjadi satu DataFrame bersih berisi 19 baris.

Sumber bacaan resmi: pandas.read_csv , pandas.read_json , dan pandas.read_sql .