Prefect

Sebelum pipeline bisa dijadwalkan, sebagian besar orchestrator menghendaki komponen server berjalan lebih dahulu. Prefect mengambil arah berbeda: flow adalah fungsi Python biasa, bisa dijalankan langsung dengan python berkas.py, dan baru dipromosikan menjadi deployment terjadwal ketika pipeline tersebut siap. Pendekatan ini menurunkan hambatan awal secara drastis — kode pipeline berjalan dulu sebagai skrip, observability dan retry menyusul tanpa menulis ulang. Versi stabil saat riset course ini ditulis adalah Prefect 3.x. Kita akan menulis flow dan task, memahami retry bawaan, lalu membedah konsep deployment dan work pool.

Flow dan Task #

Dua dekorator adalah seluruh API yang dibutuhkan: @flow menandai titik masuk workflow, @task menandai langkah yang ingin dipantau dan dicoba ulang:

# Sketsa flow Prefect 3.x.
# Diverifikasi sintaks dengan py_compile; tidak dieksekusi di bab ini.

from prefect import flow, task


@task(retries=2, retry_delay_seconds=5)
def ekstrak_pesanan() -> list[dict]:
    """Baca pesanan baru dari tabel staging (ilustrasi)."""
    return [
        {"id": 1, "pelanggan": "Budi", "total": 150000},
        {"id": 2, "pelanggan": "Siti", "total": 75000},
    ]


@task
def hitung_ringkasan(pesanan: list[dict]) -> dict:
    total = sum(p["total"] for p in pesanan)
    return {"jumlah": len(pesanan), "pendapatan": total}


@flow(name="pipeline-penjualan")
def pipeline_penjualan() -> dict:
    pesanan = ekstrak_pesanan()
    ringkasan = hitung_ringkasan(pesanan)
    return ringkasan


if __name__ == "__main__":
    pipeline_penjualan.serve(name="produksi", cron="0 6 * * *")

Mari kita telaah baris-baris kode di atas:

1. Baris 4 (import): dekorator flow dan task diimpor dari paket prefect — tanpa kelas DAG, tanpa berkas konfigurasi.
2. Baris 7-13 (ekstrak_pesanan): argumen retries=2 dan retry_delay_seconds=5 memerintahkan Prefect mengulang task hingga dua kali dengan jeda lima detik saat galat — konsep retry dan backoff yang kita bahas pada Chapter Retries dan Timeouts, di sini cukup dinyatakan lewat dekorator.
3. Baris 16-19 (hitung_ringkasan): task kedua menerima hasil task pertama sebagai argumen; Prefect mencatat ketergantungan ini dan mengalirkan nilainya secara otomatis.
4. Baris 22-26 (pipeline_penjualan): dekorator @flow membungkus fungsi menjadi workflow; isi fungsinya tetap Python biasa sehingga loop, kondisional, dan pemanggilan task lain berjalan sebagaimana mestinya.
5. Baris 29-30 (serve): menjalankan flow ini sebagai deployment bernama produksi dengan jadwal cron harian pukul 06:00; sebuah proses lokal menunggu jadwal dan mengeksekusi flow — jalur termudah menuju penjadwalan tanpa server terpisah. Menutup terminal menghentikan penjadwalan, sehingga pola ini cocok untuk pengembangan dan mesin tunggal.

Kekuatan sesungguhnya pendekatan ini ada pada workflow dinamis. Karena isi flow dieksekusi sebagai Python sungguhan, jumlah task dapat berbeda setiap run — misalnya satu task per berkas yang ditemukan di folder sumber, atau task tambahan hanya ketika validasi menemukan anomali. Pada model yang memparsing berkas tanpa mengeksekusinya, struktur semacam ini harus dinyatakan lewat pola pembuatan DAG yang tidak sederhana.

Menjalankan flow juga tidak menuntut server sejak awal. Prefect dapat berjalan sepenuhnya lokal: flow dieksekusi sebagai skrip biasa dan jejak run tersimpan di mesin tersebut. Ketika tim membutuhkan UI bersama atau riwayat run terpusat, satu perintah prefect server start menghidupkan server open-source di lokal, sementara layanan terkelola Prefect Cloud menggantikannya tanpa mengubah kode flow — peralihan di antara keduanya hanya soal konfigurasi koneksi API.

Deployment dan Work Pool #

Ketika pipeline berjalan di lebih dari satu mesin, Prefect memisahkan “siapa memicu” dari “siapa mengeksekusi” lewat tiga lapisan: deployment adalah identitas terpaket dari sebuah flow — versi kode, parameter, dan jadwalnya; work pool adalah antrean yang menampung run yang menunggu; dan worker adalah proses ringan yang menarik run dari pool lalu mengeksekusikannya:

# Sketsa deployment ke work pool — Prefect 3.x.
# Diverifikasi sintaks dengan py_compile; tidak dieksekusi di bab ini.

from pathlib import Path

from prefect import flow


@flow(name="pipeline-penjualan")
def pipeline_penjualan() -> dict:
    """Definisi flow sama seperti berkas sebelumnya (dipendekkan)."""
    return {"jumlah": 0, "pendapatan": 0}


if __name__ == "__main__":
    pipeline_penjualan.from_source(
        source=str(Path(__file__).parent),
        entrypoint="prefect_flow.py:pipeline_penjualan",
    ).deploy(
        name="produksi",
        work_pool_name="pool-produksi",
    )

Dua baris kunci pada berkas tersebut: from_source(...) menunjuk lokasi kode (repositori Git di produksi, folder lokal di pengembangan) beserta entrypoint berkas:nama_flow, dan .deploy(...) mendaftarkan deployment ke work pool bernama pool-produksi. Urutan pengoperasian work pool, secara konsep:

1. Buat work pool sekali: prefect work-pool create --type process pool-produksi.
2. Daftarkan deployment ke pool tersebut dengan menjalankan berkas di atas.
3. Jalankan worker di mesin eksekusi: prefect worker start --pool pool-produksi.
4. Picu run — dari UI, API, atau jadwal; worker mengambil run dari antrean dan mengeksekusinya.

Varian push work pool bahkan menghapus worker — run dikirim langsung ke infrastruktur terkelola seperti serverless.

Kenapa pemisahan ini penting? Karena pemicu dan eksekusi bisa gagal secara terpisah: UI dan API tetap hidup saat worker mengalami gangguan, run menunggu dengan aman di antrean, dan kapasitas eksekusi ditambah cukup dengan menjalankan worker baru tanpa menyentuh server.

Kekuatan dan Kelemahan #

Ringkasan posisi Prefect berdampingan dengan dua orchestrator sebelumnya:

Prefect paling terasa nilainya ketika tim Python-heavy membangun workflow yang bentuknya bergantung pada data, dengan kebutuhan naik kelas dari skrip lokal menjadi pipeline terjadwal tanpa menulis ulang. Namun jadwal dan event bukan satu-satunya cara pipeline dipicu — banyak sistem data modern bergerak segera setelah kejadian tertentu tiba, pola yang kita bahas pada Chapter Event-Driven Workflows. Dokumentasi lengkap tersedia di dokumentasi Prefect .