Pipeline Reusability

Toko online kita membuka kanal penjualan baru di marketplace, dan pipeline pesanannya ternyata sembilan puluh persen sama dengan pipeline kasir gudang: baca sumber, hitung total, muat ke tabel. Godaan terbesar adalah menyalin-tempel direktori pipeline lama, mengganti beberapa baris, lalu melanjukan hal lain. Godaan tersebut terasa cepat sampai kanal ketiga dibuka bulan depan dan sebuah bug di fungsi transform harus diperbaiki di tiga salinan kode yang saling mirip namun tidak identik. Desain reusable membalik situasi ini: satu kode, banyak deployment.

Tiga Pilar Desain Reusable #

Reusability tidak jatuh dari langit; ada tiga pilar yang saling menopang. Pertama, config-driven: setiap perbedaan antar deployment — lokasi sumber, nama kolom, aturan bisnis — dipindahkan dari kode ke berkas config, sehingga kode sendiri hanya menulis sekali. Kedua, fungsi generik: pembaca sumber bekerja per format, dan fungsi transform menerima aturan sebagai parameter, bukan menyimpan aturan tertulis mati di dalam tubuhnya. Ketiga, template: struktur extract → transform → load yang sama dipakai berulang, dan bagian yang berbeda cukup diisi config — deployment baru berarti config baru plus data uji, bukan proyek baru.

Kapan Generalisasi Jadi Overkill #

Generalisasi punya harga. Setiap parameter yang ditambahkan memperluas kombinasi yang harus diuji, dan abstraksi yang dibuat dari satu contoh cenderung salah bentuk: parameter yang tidak pernah dipakai, cabang logika yang tidak pernah dijalankan. Karena itu pegang rule of three: biarkan pola berulang sampai ketiga kalinya sebelum diabstraksi. Dua pipeline yang mirip masih wajar berdiri sendiri; ketika yang ketiga muncul, persamaan dan perbedaannya sudah terlihat jelas dan abstraksi bisa dibuat pada titik yang tepat.

Library Internal vs Copy-Paste #

Untuk logika yang memang dipakai bersama, pilihannya dua: menyalin-tempel kode, atau mengangkatnya menjadi library internal yang berversi. Perbandingannya lugas:

Implementasi: Satu Kode, Dua Deployment #

Kita uji pilar-pilar tersebut dengan dua deployment nyata: kasir gudang yang mengekspor JSONL dengan nama kolom standar, dan marketplace yang mengirim CSV dengan nama kolom berbeda (nomor, jumlah, satuan) serta tanpa kolom diskon. Seluruh perbedaan hidup di dua berkas config berikut.

{
  "deployment": "pos-gudang",
  "basisdata": "toko_gabungan.db",
  "tabel_target": "pesanan_pos",
  "sumber": {"tipe": "jsonl", "berkas": "pesanan_gudang.jsonl"},
  "pemetaan": {
    "pesanan_id": "pesanan_id", "waktu": "waktu", "pelanggan": "pelanggan",
    "produk": "produk", "qty": "qty", "harga": "harga", "diskon": "diskon"
  },
  "tipe": {"pesanan_id": "int", "qty": "int", "harga": "int", "diskon": "int"},
  "aturan": {"ambang_besar": 500000}
}

{
  "deployment": "marketplace",
  "basisdata": "toko_gabungan.db",
  "tabel_target": "pesanan_marketplace",
  "sumber": {"tipe": "csv", "berkas": "pesanan_marketplace.csv"},
  "pemetaan": {
    "pesanan_id": "nomor", "waktu": "tanggal", "pelanggan": "pembeli",
    "produk": "barang", "qty": "jumlah", "harga": "satuan"
  },
  "tipe": {"pesanan_id": "int", "qty": "int", "harga": "int"},
  "aturan": {"ambang_besar": 300000}
}

Kode pipeline-nya satu, tanpa satu pun nama deployment tertulis di dalamnya:

"""Pipeline pesanan config-driven: satu kode untuk banyak deployment."""
import csv
import json
import sqlite3
import sys
from pathlib import Path


def baca_jsonl(berkas, pemetaan):
    for baris in berkas.read_text().splitlines():
        data = json.loads(baris)
        yield {target: data[sumber] for target, sumber in pemetaan.items()}


def baca_csv(berkas, pemetaan):
    with berkas.open(newline="") as f:
        for data in csv.DictReader(f):
            yield {target: data[sumber] for target, sumber in pemetaan.items()}


PEMBACA = {"jsonl": baca_jsonl, "csv": baca_csv}
KONVERTER = {"int": int, "text": str}


def konversi(baris, tipe_kolom):
    for kolom, tipe in tipe_kolom.items():
        if baris.get(kolom) is not None:
            baris[kolom] = KONVERTER[tipe](baris[kolom])
    return baris


def transform(baris, aturan):
    if baris["qty"] <= 0:
        raise ValueError(f"qty tidak valid: {baris['qty']}")
    total = baris["qty"] * baris["harga"] - (baris.get("diskon") or 0)
    segmen = "besar" if total >= aturan.get("ambang_besar", 500000) else "kecil"
    return (baris["pesanan_id"], baris["waktu"], baris["pelanggan"],
            baris["produk"], baris["qty"], baris["harga"], total, segmen)


def jalankan(berkas_config):
    config = json.loads(berkas_config.read_text())
    sumber = Path(config["sumber"]["berkas"])
    pembaca = PEMBACA[config["sumber"]["tipe"]]
    koneksi = sqlite3.connect(config["basisdata"])
    koneksi.execute(
        f"""CREATE TABLE IF NOT EXISTS {config['tabel_target']} (
                pesanan_id INTEGER PRIMARY KEY, waktu TEXT NOT NULL,
                pelanggan TEXT NOT NULL, produk TEXT NOT NULL,
                qty INTEGER NOT NULL, harga INTEGER NOT NULL,
                total INTEGER NOT NULL, segmen TEXT NOT NULL)""")
    jumlah = 0
    for baris in pembaca(sumber, config["pemetaan"]):
        baris = konversi(baris, config.get("tipe", {}))
        koneksi.execute(
            f"INSERT OR REPLACE INTO {config['tabel_target']} VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            transform(baris, config.get("aturan", {})))
        jumlah += 1
    koneksi.commit()
    print(f"=== Deployment {config['deployment']} ({berkas_config.name}) ===")
    print(f"  sumber {config['sumber']['tipe']} {sumber.name}: {jumlah} baris dibaca")
    for segmen, isi in koneksi.execute(
        f"SELECT segmen, COUNT(*) FROM {config['tabel_target']} GROUP BY segmen"
    ):
        print(f"  tabel {config['tabel_target']}: segmen {segmen} {isi} baris")
    koneksi.close()


if __name__ == "__main__":
    for nama_config in sys.argv[1:]:
        jalankan(Path(nama_config))

Kedua deployment dijalankan oleh perintah yang sama — cukup daftarkan kedua berkas config:

python3 pipeline_configurable.py config_gudang.json config_marketplace.json

Output yang Dihasilkan:

=== Deployment pos-gudang (config_gudang.json) ===
  sumber jsonl pesanan_gudang.jsonl: 5 baris dibaca
  tabel pesanan_pos: segmen besar 2 baris
  tabel pesanan_pos: segmen kecil 3 baris
=== Deployment marketplace (config_marketplace.json) ===
  sumber csv pesanan_marketplace.csv: 5 baris dibaca
  tabel pesanan_marketplace: segmen besar 3 baris
  tabel pesanan_marketplace: segmen kecil 2 baris

Mari kita telaah baris-baris kode di atas:

1. Baris 9-18 (baca_jsonl, baca_csv): dua pembaca generik — keduanya menerima pemetaan dan menghasilkan baris dengan nama kolom standar. Perbedaan format dan nama kolom sumber berhenti di fungsi ini.
2. Baris 21-22 (PEMBACA, KONVERTER): registry memetakan tipe sumber ke fungsi pembaca dan nama tipe ke fungsi konversi, menggantikan rantai if-else di badan pipeline.
3. Baris 25-29 (konversi): CSV selalu memberi nilai string, sehingga tipe kolom dinyatakan di config dan konversi menyamakan bentuk data lintas sumber.
4. Baris 32-38 (transform): aturan bisnis datang dari parameter aturan — ambang segmen besar ditentukan config, bukan tertulis mati; diskon bersifat opsional sehingga sumber tanpa diskon tetap sah.
5. Baris 41-59 (jalankan): seluruh identitas deployment dibaca dari config — berkas sumber, basisdata, tabel target, pemetaan, dan aturan — lalu pipeline menulis dengan kontrak output delapan kolom yang sama untuk semua deployment.
6. Baris 69-71 (__main__): satu kode melayani daftar config sepanjang apa pun; deployment ketiga tinggal menambah satu berkas config.

Hasilnya dua tabel dengan kontrak kolom identik — pesanan_pos dan pesanan_marketplace sama-sama delapan kolom — sehingga konsumen downstream bisa menggabungkan keduanya tanpa peduli asal kanal. Bug pada perhitungan total cukup diperbaiki sekali di transform, dan kedua deployment langsung memperoleh perbaikan pada run berikutnya.

Satu kode yang melayani banyak deployment membuat pertumbuhan volume terasa cepat: setiap kanal baru menambah beban pada kode yang sama. Ketika durasi run mulai memanjang, saatnya mengukur dan mengoptimalkan — pokok bahasan Chapter Pipeline Performance.