Desain pipeline penjualan harian pada Chapter Merancang Data Pipeline sudah memilih pola batch: data diolah sekali per hari pada jadwal tetap. Pola ini merupakan fondasi seluruh pola lain karena konsepnya paling sederhana: pipeline membaca himpunan data yang sudah selesai (bounded), mengubahnya, lalu menulis hasilnya ke tujuan. Pada chapter ini kita bedah karakteristik batch, aturan jendela eksekusi, dan pola read-process-write per partisi, kemudian merakitnya menjadi pipeline batch harian TokoKita yang benar-benar berjalan.

Karakteristik Pipeline Batch #

Pipeline batch memproses data dalam jumlah besar pada satu waktu terjadwal, misalnya sekali per hari atau sekali per jam. Data yang dibaca berstatus selesai: semua pesanan 15 Januari sudah final ketika pipeline berjalan pada dini hari 16 Januari, sehingga pipeline tidak perlu berurusan dengan data yang masih berubah. Konsekuensinya, latensi batch dihitung dalam jam, bukan detik, dan itu bukan masalah untuk laporan harian.

Sisi baiknya, batch paling mudah diuji dan dijalankan ulang. Input berupa berkas atau tabel yang tidak berubah, sehingga hasil eksekusi ulang dapat diprediksi. Ketika tim bisnis menemukan galat pada laporan, kita cukup menjalankan ulang partisi tanggal tersebut tanpa menyentuh partisi lain.

Jendela Eksekusi #

Jendela eksekusi adalah rentang waktu yang disediakan untuk pipeline selesai sebelum hasilnya dibutuhkan. TokoKita menetapkan jendela 01:30 sampai 02:30 dengan durasi nyata di bawah 45 menit, karena pukul 06:00 laporan harus sudah siap di hadapan tim bisnis. Volume sekitar 12 ribu baris per hari membuat jendela tersebut longgar, tetapi struktur jendela tetap wajib: begitu volume naik sepuluh kali, jendela yang tadinya longgar bisa berubah padat.

Jendela juga menentukan protokol kegagalan. Pipeline yang gagal pukul 02:00 masih punya empat jam untuk dijalankan ulang sebelum SLA terlampaui. Sebaliknya, pipeline yang baru selesai pukul 05:55 tidak menyisakan ruang pulih sama sekali.

Pola Read-Process-Write per Partisi #

Pipeline batch TokoKita tidak membaca semua tanggal dalam satu proses raksasa. Setiap tanggal menjadi satu partisi, dan pipeline menjalankan siklus read-process-write untuk tiap partisi secara terpisah: baca orders_2026-01-05.csv, proses, tulis ke tabel orders; lalu lanjut ke 2026-01-06, dan seterusnya. Pemisahan per partisi memberikan tiga keuntungan: kegagalan di satu partisi tidak menular ke partisi lain, eksekusi ulang cukup pada partisi yang bermasalah, dan tiap partisi dapat diproses paralel di masa depan tanpa mengubah kode.

Contoh: Pipeline Batch Harian TokoKita #

Sumber kita adalah berkas CSV di data/orders, satu berkas per tanggal. Berikut isi orders_2026-01-05.csv sebagai contoh; dua berkas lainnya mengikuti struktur yang sama untuk tanggal 06 dan 07.

text 
order_id,pelanggan,email,tanggal,status,total
1001,Budi,[email protected],2026-01-05,Dikirim,450000
1002,Siti,[email protected],2026-01-05,diproses,275000
1003,Dewi,[email protected],2026-01-05,batal,300000
1004,Andi,[email protected],2026-01-05,dikirim,825000
1005,Rina,[email protected],2026-01-05,dikirim,400000

Pipeline membaca partisi tersebut, membuang pesanan batal, menyeragamkan status dan email ke huruf kecil, lalu memuatnya ke tabel orders di SQLite. Setelah seluruh partisi masuk, tabel ringkasan_harian dihitung ulang penuh dari tabel orders, sehingga ringkasan selalu konsisten dengan isinya.

python 
"""Pipeline batch harian TokoKita: ekstrak CSV -> transform -> load SQLite."""
import csv
import sqlite3
from pathlib import Path

SUMBER = Path("data/orders")
WAREHOUSE = Path("out/warehouse.db")


def baca_partisi(tanggal):
    with (SUMBER / f"orders_{tanggal}.csv").open() as f:
        return list(csv.DictReader(f))


def transformasi(baris):
    hasil = []
    for b in baris:
        if b["status"].lower() == "batal":
            continue
        hasil.append((int(b["order_id"]), b["pelanggan"], b["email"].lower(),
                      b["tanggal"], b["status"].lower(), int(b["total"])))
    return hasil


def muat_partisi(koneksi, tanggal, baris):
    koneksi.execute("DELETE FROM orders WHERE tanggal = ?", (tanggal,))
    koneksi.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?)", baris)
    koneksi.commit()


def format_rp(nilai):
    return f"Rp{nilai:,}".replace(",", ".")


WAREHOUSE.parent.mkdir(parents=True, exist_ok=True)
koneksi = sqlite3.connect(WAREHOUSE)
koneksi.executescript("""
CREATE TABLE IF NOT EXISTS orders (
    order_id INTEGER PRIMARY KEY, pelanggan TEXT, email TEXT,
    tanggal TEXT, status TEXT, total INTEGER);
CREATE TABLE IF NOT EXISTS ringkasan_harian (
    tanggal TEXT PRIMARY KEY, jumlah_pesanan INTEGER, total_pendapatan INTEGER);
""")

partisi = sorted(p.stem.removeprefix("orders_") for p in SUMBER.glob("orders_*.csv"))
total_baris = 0
for tanggal in partisi:
    dibaca = baca_partisi(tanggal)
    valid = transformasi(dibaca)
    muat_partisi(koneksi, tanggal, valid)
    total_baris += len(valid)
    print(f"partisi {tanggal}: {len(dibaca)} dibaca, {len(valid)} valid")

print(f"load selesai: {total_baris} baris di tabel orders")
koneksi.execute("DELETE FROM ringkasan_harian")
koneksi.execute("""
    INSERT INTO ringkasan_harian
    SELECT tanggal, COUNT(*), SUM(total) FROM orders GROUP BY tanggal
""")
koneksi.commit()
print("ringkasan_harian dihitung ulang dari tabel orders:")
for tanggal, jumlah, total in koneksi.execute(
        "SELECT * FROM ringkasan_harian ORDER BY tanggal"):
    print(f"  {tanggal} | {jumlah} pesanan | {format_rp(total)}")
koneksi.close()

Output yang Dihasilkan:

text 
partisi 2026-01-05: 5 dibaca, 4 valid
partisi 2026-01-06: 6 dibaca, 5 valid
partisi 2026-01-07: 4 dibaca, 4 valid
load selesai: 13 baris di tabel orders
ringkasan_harian dihitung ulang dari tabel orders:
  2026-01-05 | 4 pesanan | Rp1.950.000
  2026-01-06 | 5 pesanan | Rp2.730.000
  2026-01-07 | 4 pesanan | Rp2.010.000

Mari kita telaah baris-per-baris kode di atas:

  1. Baris 10-12 (baca_partisi): fungsi membuka berkas partisi sesuai tanggal dan mengembalikan daftar baris dalam bentuk dictionary per kolom.
  2. Baris 15-22 (transformasi): baris berstatus batal dibuang, email dan status diseragamkan ke huruf kecil, dan total dikonversi menjadi integer sebelum dikemas sebagai tuple.
  3. Baris 25-28 (muat_partisi): klausa DELETE ... WHERE tanggal = ? menghapus partisi lama di tabel orders terlebih dahulu, lalu baris baru disisipkan. Kombinasi hapus-lalu-sisip membuat eksekusi ulang pada partisi yang sama selalu menghasilkan isi akhir yang sama.
  4. Baris 45: daftar partisi diambil dari pola nama berkas orders_*.csv, lalu diurutkan sehingga pipeline selalu memproses tanggal secara kronologis.
  5. Baris 55-59: tabel ringkasan_harian dibersihkan dan dihitung ulang seluruhnya lewat satu query GROUP BY, bukan ditambah sedikit demi sedikit.

Perhatikan baris keluaran untuk partisi 2026-01-06: 6 baris dibaca menjadi 5 valid, karena satu pesanan berstatus batal dibuang saat transformasi. Ringkasan harian kemudian cocok dengan tabel orders: 4 + 5 + 4 = 13 baris menghasilkan tiga baris ringkasan.

Dokumentasi modul csv tersedia di dokumentasi Python , dan perilaku INSERT serta GROUP BY yang dipakai di atas terdokumentasi di dokumentasi SQLite . Pipeline batch ini membaca ulang seluruh partisi setiap kali berjalan, dan itu wajar untuk tiga hari data. Ketika riwayat menumpuk hingga berbulan-bulan, membaca ulang data lama yang tidak berubah menjadi pemborosan, dan di titik itu kita membutuhkan Chapter Pipeline Incremental.