Pemuatan adalah tahap yang paling sering diulang: berkas dikoreksi di sumber, diekspor ulang, lalu pipeline dijalankan lagi. Pada bagian ini kita menulis pemuatan agar pengulangan seperti itu aman lewat UPSERT, membagi pemuatan menjadi batch agar tidak menahan seluruh data di memori, dan menyiapkan loader khusus PostgreSQL. SQLite dijalankan nyata di lingkungan tutorial sebagai kembaran lokal; loader PostgreSQL diverifikasi sampai kompilasi dan dijelaskan perilakunya dalam prosa.

Upsert sebagai Dasar #

Satu pernyataan SQL di src/load.py menjadi dasar seluruh pemuatan:

sql 
INSERT INTO pesanan (order_id, tanggal, pelanggan, email, produk, kategori, jumlah, harga, status)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT (order_id) DO UPDATE SET
    tanggal  = excluded.tanggal,
    pelanggan = excluded.pelanggan,
    email    = excluded.email,
    produk   = excluded.produk,
    kategori = excluded.kategori,
    jumlah   = excluded.jumlah,
    harga    = excluded.harga,
    status   = excluded.status

Mari kita telaah pernyataan ini:

  1. INSERT INTO pesanan (...): menyisipkan satu baris dengan sembilan kolom; kolom id dan dibuat_pada tidak diisi karena digenerasi tabel.
  2. ON CONFLICT (order_id): bila order_id sudah ada di tabel, penyisipan tidak galat; konfliknya ditangani pernyataan yang sama.
  3. DO UPDATE SET ... = excluded.kolom: baris lama diperbarui dengan nilai baru. Kata kunci excluded merujuk pada baris yang gagal disisipkan tadi.

Hasilnya: memuat 189 baris dua kali masih menghasilkan 189 baris. Sifat ini memenuhi kriteria bisa dijalankan ulang dari Bagian 1 dan adalah wujud praktis dari Idempotency dan Retries.

Pemuatan Per Batch #

Fungsi muat_batch membagi daftar baris menjadi potongan dan memakai executemany untuk setiap potongan:

python 
def muat_batch(conn: sqlite3.Connection, baris: list[dict], ukuran_batch: int = 100) -> int:
    """Muat baris dengan executemany per potongan; kembalikan jumlah baris diproses.

    INSERT ... ON CONFLICT DO UPDATE membuat pemuatan idempotent: baris lama
    diperbarui, baris baru disisipkan, tidak ada duplikat kunci.
    """
    terproses = 0
    for mulai in range(0, len(baris), ukuran_batch):
        potongan = baris[mulai : mulai + ukuran_batch]
        conn.executemany(SQL_MUAT, [tuple(b[k] for k in KOLOM_ISI) for b in potongan])
        terproses += len(potongan)
    conn.commit()
    return terproses

Batch 100 baris berarti satu putaran jaringan dan satu commit per seratus baris. Di PostgreSQL sungguhan, teknik sejenisnya adalah COPY untuk data yang benar-benar baru; INSERT ... ON CONFLICT kita pilih karena pipeline ini menuntut pembaruan dan penyisipan dalam satu langkah.

Idempotensi dalam Praktik #

Skrip berikut dijalankan nyata. Tiga baris dimuat, satu harga dikoreksi, lalu seluruh berkas dimuat ulang:

python 
n1 = load.muat_batch(koneksi, contoh)
print(f"muat pertama : {n1} baris, tabel berisi {load.hitung_baris(koneksi)}")

contoh[1]["harga"] = 250000
n2 = load.muat_batch(koneksi, contoh)
print(f"muat kedua   : {n2} baris diproses, tabel tetap berisi {load.hitung_baris(koneksi)}")

Output yang Dihasilkan:

text 
muat pertama : 3 baris, tabel berisi 3
muat kedua   : 3 baris diproses, tabel tetap berisi 3
harga ORD-2024-0002 sekarang: 250000
constraint menolak: CHECK constraint failed: jumlah > 0

Tiga baris terakhir pada output membuktikan dua hal sekaligus. Pertama, pemuatan kedua memproses 3 baris tetapi tabel tetap berisi 3: upsert menimpa, bukan menambah. Kedua, percobaan menyisipkan jumlah = 0 ditolak constraint CHECK dari skema Bagian 2; database menutup celah yang mungkin lolos dari Python.

Bila Pemuatan Terhenti di Tengah #

Batch yang sudah dicommit tidak diurungkan bila batch berikutnya gagal. muat_batch melakukan conn.commit() per potongan, jadi kegagalan di potongan ketiga meninggalkan dua potongan pertama di tabel. Dalam desain ini bukan bencana, melainkan keadaan yang dapat disembuhkan: karena setiap baris dimuat lewat upsert, menjalankan ulang pipeline akan menulis ulang baris-baris tersebut dan melanjutkan sisanya, bukan menggandakan apa pun.

Pilihan ini menukar all-or-nothing dengan convergent on retry. Untuk pipeline yang selalu bisa dijalankan ulang, pilihan kedua lebih sederhana dan lebih mudah diuji; pembahasan kompromi lengkapnya ada di Transaksi Database.

Loader Khusus PostgreSQL #

Berkas src/loader_postgres.py memakai psycopg 3 dan idiom khas PostgreSQL:

python 
SQL_MUAT_POSTGRES = """
INSERT INTO pesanan (order_id, tanggal, pelanggan, email, produk, kategori, jumlah, harga, status)
VALUES (%(order_id)s, %(tanggal)s, %(pelanggan)s, %(email)s, %(produk)s, %(kategori)s, %(jumlah)s, %(harga)s, %(status)s)
ON CONFLICT (order_id) DO UPDATE SET
    tanggal   = EXCLUDED.tanggal,
    pelanggan = EXCLUDED.pelanggan,
    email     = EXCLUDED.email,
    produk    = EXCLUDED.produk,
    kategori  = EXCLUDED.kategori,
    jumlah    = EXCLUDED.jumlah,
    harga     = EXCLUDED.harga,
    status    = EXCLUDED.status
""".strip()

SQL_MUAT_RETURNING = SQL_MUAT_POSTGRES + "\nRETURNING id, order_id"

Fungsi muat_batch_postgres menyalin perilaku muat_batch dengan penanda tempat %(kolom)s khas psycopg, dan ambil_atau_perbarui menambahkan RETURNING id, order_id sehingga satu bolak-balik ke server sudah mengembalikan kunci baris hasil, tanpa query tambahan.

Parameter Koneksi Terpisah dari Kode #

Loader PostgreSQL membaca parameternya dari config/database.yaml, yang salinan contohnya sudah kita lihat di Bagian 2. Fungsi connect_from_yaml tinggal membuka berkas tersebut dan meneruskan isinya ke psycopg.connect. Kredensial tidak pernah ditulis di kode dan tidak pernah dicatat di log.

Materi pendalaman untuk pola ini ada di Pemuatan Data, Transaksi Database, dan Connection Pooling. Rujukan resmi: dokumentasi UPSERT SQLite dan dokumentasi PostgreSQL .

Baris bersih kini punya mekanisme masuk yang aman diulang. Satu ancaman masih tersisa: pesanan yang sama muncul dua kali di dalam data itu sendiri. Lanjutkan ke Bagian 6: Menangani Data Duplikat.