Bagian 9: Menjalankan Pipeline

Semua lapisan telah dibangun dan lulus pengujian; saatnya menjalankan keseluruhan pipeline pada data asli dan membaca hasilnya dengan kritis. Pada bagian ini kita menjalankan pipeline dua kali untuk membuktikan sifat idempotent, memverifikasi isi tabel lewat query, lalu menimbang kembali setiap kriteria keberhasilan yang ditetapkan di Bagian 1. Bagian ini sekaligus menutup proyek.

Menjalankan dari Root Proyek #

Titik masuk CLI membaca config/pipeline.yaml sebagai nilai bawaan, dan opsi baris perintah menimpa nilai tersebut:

Perintah	Arti
`python3 -m src.pipeline`	jalankan dengan konfigurasi bawaan
`python3 -m src.pipeline --db out/uji.db`	arahkan basisdata keluaran
`python3 -m src.pipeline --level WARNING`	hanya catat level WARNING ke atas
`python3 -m src.pipeline --pola "pesanan_2024-06.csv"`	proses satu berkas saja

Eksekusi Nyata #

Perintah berikut dijalankan apa adanya tanpa opsi tambahan:

bash

python3 -m src.pipeline

Output yang Dihasilkan:

text
2026-10-17 07:42:21 | INFO     | Memulai pipeline: 2 berkas sumber
2026-10-17 07:42:21 | WARNING  | pesanan_2024-06.csv: baris 9 ditolak (email_tidak_valid): email harus berformat [email protected]
2026-10-17 07:42:21 | WARNING  | pesanan_2024-06.csv: baris 14 ditolak (jumlah_tidak_positif): jumlah tidak boleh negatif
2026-10-17 07:42:21 | WARNING  | pesanan_2024-06.csv: baris 20 ditolak (harga_kosong): harga tidak boleh kosong
2026-10-17 07:42:21 | WARNING  | pesanan_2024-06.csv: baris 25 ditolak (tanggal_tidak_valid): tanggal harus tanggal nyata berformat YYYY-MM-DD
2026-10-17 07:42:21 | WARNING  | pesanan_2024-06.csv: baris 32 ditolak (order_id_tidak_valid): pola order_id harus ORD-YYYY-NNNN
2026-10-17 07:42:21 | WARNING  | pesanan_2024-06.csv: baris 38 ditolak (kategori_tidak_dikenal): kategori harus salah satu dari: buku, elektronik, fashion, makanan, olahraga
2026-10-17 07:42:21 | WARNING  | pesanan_2024-06.csv: baris 43 ditolak (pelanggan_kosong): nama pelanggan tidak boleh kosong
2026-10-17 07:42:21 | WARNING  | pesanan_2024-06.csv: baris 49 ditolak (tanggal_tidak_valid): tanggal harus tanggal nyata berformat YYYY-MM-DD
2026-10-17 07:42:21 | WARNING  | pesanan_2024-06.csv: baris 54 ditolak (jumlah_tidak_positif): jumlah minimal 1
2026-10-17 07:42:21 | WARNING  | pesanan_2024-06.csv: baris 60 ditolak (email_kosong): email tidak boleh kosong
2026-10-17 07:42:21 | WARNING  | pesanan_2024-06.csv: baris 65 ditolak (produk_kosong): nama produk tidak boleh kosong
2026-10-17 07:42:21 | WARNING  | pesanan_2024-06.csv: baris 71 ditolak (harga_tidak_valid): harga harus angka >= 0, boleh berpemisah ribuan (210.000)
2026-10-17 07:42:21 | WARNING  | pesanan_2024-06.csv: baris 76 ditolak (status_tidak_dikenal): status harus salah satu dari: dibatalkan, dikirim, diproses, selesai
2026-10-17 07:42:21 | WARNING  | pesanan_2024-06.csv: baris 82 ditolak (tanggal_kosong): tanggal tidak boleh kosong
2026-10-17 07:42:21 | WARNING  | pesanan_2024-06.csv: baris 87 ditolak (order_id_kosong): order_id tidak boleh kosong
2026-10-17 07:42:21 | WARNING  | pesanan_2024-06.csv: baris 92 ditolak (harga_negatif): harga tidak boleh negatif
2026-10-17 07:42:21 | INFO     | Validasi pesanan_2024-06.csv: 107 baris dibaca, 16 ditolak, 91 lolos
2026-10-17 07:42:21 | WARNING  | pesanan_2024-07.csv: baris 11 ditolak (email_tidak_valid): email harus berformat [email protected]
2026-10-17 07:42:21 | WARNING  | pesanan_2024-07.csv: baris 22 ditolak (jumlah_tidak_valid): jumlah harus angka bulat
2026-10-17 07:42:21 | WARNING  | pesanan_2024-07.csv: baris 35 ditolak (tanggal_tidak_valid): tanggal harus tanggal nyata berformat YYYY-MM-DD
2026-10-17 07:42:21 | WARNING  | pesanan_2024-07.csv: baris 57 ditolak (kategori_kosong): kategori tidak boleh kosong
2026-10-17 07:42:21 | WARNING  | pesanan_2024-07.csv: baris 72 ditolak (status_kosong): status tidak boleh kosong
2026-10-17 07:42:21 | INFO     | Validasi pesanan_2024-07.csv: 111 baris dibaca, 5 ditolak, 106 lolos
2026-10-17 07:42:21 | INFO     | Pembersihan: 197 baris bersih dihasilkan
2026-10-17 07:42:21 | WARNING  | Deduplikasi: 7 duplikat persis dan 1 konflik kunci dibuang
2026-10-17 07:42:21 | INFO     | Deduplikasi: 197 baris masuk, 189 baris unik keluar
2026-10-17 07:42:21 | INFO     | CSV bersih ditulis: data/processed/pesanan_bersih.csv (189 baris)
2026-10-17 07:42:21 | INFO     | Pemuatan: 189 baris dimuat ke data/processed/tokokita.db (tabel pesanan)
2026-10-17 07:42:21 | INFO     | Verifikasi: tabel pesanan berisi 189 baris
2026-10-17 07:42:21 | INFO     | Pipeline selesai: dibaca 218, ditolak 21, unik 189, di tabel 189

Waktu pada kolom pertama adalah waktu eksekusi nyata; setiap eksekusi mencatat waktu yang berbeda. Sisa angkanya tetap.

Membaca Angka Ringkasan #

Aritmetika pada baris terakhir dapat diperiksa ulang dari sumbernya:

Tahap	Angka	Sumber
Dibaca	218	107 baris Juni + 111 baris Juli
Ditolak	21	16 temuan Juni + 5 temuan Juli
Bersih	197	91 lolos Juni + 106 lolos Juli
Duplikat persis	7	2 dalam Juni + 2 dalam Juli + 3 lintas berkas
Konflik	1	`ORD-2024-0046` versi Juli
Unik	189	197 dikurangi 7 duplikat dan 1 konflik
Di tabel	189	verifikasi `COUNT(*)` setelah pemuatan

Kecocokan antara 189 baris unik dan 189 baris di tabel bukan kebetulan; angka terakhir diambil dari query verifikasi ke basisdata, bukan dari variabel Python. Bila keduanya berbeda, pipeline wajib dihentikan dan diperiksa.

Idempotensi: Menjalankan Dua Kali #

Kriteria bisa dijalankan ulang diuji dengan menjalankan pipeline kedua kali pada basisdata yang sama:

bash

python3 -m src.pipeline --log data/processed/pipeline2.log | tail -2

Output yang Dihasilkan:

text

2026-10-17 07:43:27 | INFO     | Verifikasi: tabel pesanan berisi 189 baris
2026-10-17 07:43:27 | INFO     | Pipeline selesai: dibaca 218, ditolak 21, unik 189, di tabel 189

Eksekusi kedua memproses 189 baris yang sama, tetapi tabel tetap berisi 189 baris. Upsert dari Bagian 5 mengubah baris lama menjadi dirinya sendiri; tidak ada satu baris pun bertambah.

Verifikasi Isi Tabel #

Tim analitik akan bertanya hal yang lebih tajam dari jumlah baris. Skrip berikut dijalankan nyata untuk menjawab tiga pertanyaan itu:

python
print(f"total baris   : {load.hitung_baris(koneksi)}")
for status, jumlah in load.ringkasan_status(koneksi):
    print(f"  {status:12s} {jumlah}")
for pelanggan, pesanan, total in load.top_pelanggan(koneksi, batas=5):
    print(f"  {pelanggan:16s} {pesanan:3d} pesanan  total Rp{total:,}")

Output yang Dihasilkan:

text
total baris   : 189
ringkasan status:
  Dibatalkan   17
  Dikirim      44
  Diproses     30
  Selesai      98
top 5 pelanggan menurut nilai pesanan:
  Agus Pratama       6 pesanan  total Rp17.085.000
  Yusuf Maulana      9 pesanan  total Rp13.020.000
  Maya Putri        12 pesanan  total Rp12.983.000
  Bambang Setiawan  11 pesanan  total Rp12.947.000
  Indah Permata      9 pesanan  total Rp12.735.000
contoh baris pertama:
  ORD-2024-0001 | 2024-06-19 | Dedi Kurniawan   | Powerbank 10000 mAh | 3 x Rp250.000 | Selesai
  ORD-2024-0002 | 2024-06-03 | Wulan Sari       | Tas Ransel         | 5 x Rp320.000 | Diproses
  ORD-2024-0003 | 2024-06-28 | Hendra Gunawan   | Matras Yoga        | 4 x Rp210.000 | Dikirim
kebijakan konflik ORD-2024-0046: versi Juni dipertahankan (jumlah=3, status=Dikirim)

Status tampil sebagai empat kelompok, bukan delapan kelompok terpecah huruf besar-kecil, yang berarti pembersihan benar-benar bekerja. Baris terakhir menegaskan kebijakan konflik: versi Juni yang dipertahankan, sesuai janji Bagian 6.

Tinjauan Kriteria Keberhasilan #

Kriteria dari Bagian 1 kini dapat ditutup satu per satu:

Kriteria	Bukti
0 baris ganda di tabel	`COUNT(*)` 189; kunci `order_id UNIQUE`
Baris galat ditolak dengan catatan	21 baris `WARNING` berisi nomor baris dan alasan
Nilai konsisten	4 kelompok status, bukan 8
Tipe data benar	`jumlah` dan `harga` integer di skema
Bisa dijalankan ulang	eksekusi kedua: tabel tetap 189 baris
Terlacak	31 baris log di `data/processed/pipeline.log`
Dapat diverifikasi	angka pipeline cocok dengan query tabel

Coba Selanjutnya #

Beberapa arah pengembangan yang pas setelah proyek ini:

tambahkan kategori baru, misalnya Peralatan, lalu perbarui KATEGORI, KATEGORI_CANONICAL, dan pengujianya;
ubah kebijakan konflik menjadi kemunculan terakhir, lalu bandingkan hasil tabelnya;
pasang loader PostgreSQL sungguhan: siapkan server, jalankan config/schema.sql, salin database.example.yaml, lalu ganti pemanggilan load dengan loader_postgres;
tambahkan ringkasan omzet per kategori pada tahap verifikasi;
jalankan pipeline lewat penjadwal dan kirim ringkasannya, seperti dibahas di Scheduling dan Production Runs.

Rujukan teknis penutup: dokumentasi PostgreSQL dan dokumentasi pytest .

Dari sini, langkah alami berikutnya adalah berhenti bergantung pada berkas: banyak sumber data modern hanya tersedia lewat API. Pada proyek berikutnya kita membangun ingestion dari REST API dengan pagination, rate limit, dan retry. Lanjutkan ke Proyek: Pipeline Ingestion API.