Batch Feature Computation

Tidak semua fitur harus tersedia dalam milidetik. Fitur RFM 30 hari, frekuensi belanja mingguan, atau rata-rata nilai pesanan tidak berubah maknanya bila dihitung sekali sehari — dan menghitungnya terjadwal jauh lebih murah serta lebih mudah diaudit daripada menghitungnya real-time. Chapter ini membangun pola batch feature computation yang menjadi pola dasar kebanyakan feature pipeline: job harian membaca partisi sumber, menghitung fitur dengan satu query SQL, menulis snapshot ber-partisi, dan tetap benar ketika data lama datang terlambat.

Pola Harian: Partisi Sumber, Snapshot Fitur #

Struktur datanya mengikuti kebiasaan lakehouse: tabel orders hidup sebagai partisi harian tanggal=YYYY-MM-DD berformat Parquet , dan tabel fitur hidup sebagai partisi snapshot_date. Dua fungsi berikut menjadi intinya — blok memotong bagian pembuatan 7 pesanan contoh dan skema PyArrow pada berkas scratch:

python
def tulis_partisi(pesanan, akar):
    """Tulis pesanan ke partisi harian; gabungkan bila partisi sudah ada."""
    df = pd.DataFrame(pesanan, columns=["order_id", "customer_id", "order_ts", "total"])
    for tanggal, grp in df.groupby(df["order_ts"].dt.date):
        folder = akar / f"tanggal={tanggal:%Y-%m-%d}"
        folder.mkdir(parents=True, exist_ok=True)
        target = folder / "part.parquet"
        if target.exists():
            # partisi lama digabung, bukan ditimpa -- menimpa akan membuang data
            lama = pq.read_table(target).to_pandas()
            grp = pd.concat([lama, grp]).drop_duplicates("order_id", keep="last")
        pq.write_table(pa.Table.from_pandas(grp, schema=SKEMA), target)

python
def hitung_fitur(snapshot, akar=ORDERS_DIR):
    """Fitur frek_belanja_7d + total_belanja_7d per pelanggan pada tanggal snapshot."""
    akhir = datetime.combine(snapshot + timedelta(days=1), time.min)  # eksklusif
    mulai = akhir - timedelta(days=JENDELA)
    return duckdb.sql(
        f"""
        SELECT customer_id,
               count(*)   AS frek_belanja_7d,
               sum(total) AS total_belanja_7d
        FROM read_parquet('{akar}/**/*.parquet')
        WHERE order_ts >= TIMESTAMP '{mulai:%Y-%m-%d %H:%M:%S}'
          AND order_ts <  TIMESTAMP '{akhir:%Y-%m-%d %H:%M:%S}'
        GROUP BY customer_id
        ORDER BY customer_id
        """
    ).df().assign(snapshot_date=str(snapshot))

Mari kita telaah bagian penting kode di atas:

Baris 10-11 (lama = pq.read_table ...): bila partisi sudah ada, pesanan baru digabung dengan isi lama lalu drop_duplicates("order_id", keep="last") menjadikan penulisan idempotent — menulis ulang data yang sama tidak menduplikasi baris.
Baris 12 (pq.write_table): hasil gabungan ditulis dengan skema eksplisit agar tipe kolom tidak bergeser antar penulisan.
Baris 15-16 (akhir, mulai): jendela 7 hari berakhir tepat tengah malam setelah tanggal snapshot dengan batas atas eksklusif — konvensi interval yang sama dengan chapter Feature Pipeline.
Baris 21 (read_parquet('{akar}/**/*.parquet')): satu glob membaca seluruh partisi, dan filter waktu di WHERE memastikan hanya jendela relevan yang dihitung — pola pushdown filter bawaan DuckDB .

Bagian utama skrip menjalankan pola hariannya — membersihkan direktori agar run selalu mulai dari kondisi sama, menulis pesanan, menghitung fitur untuk snapshot 1 sampai 5 November, lalu menyimpan hasilnya sebagai partisi snapshot_date, sebuah offline store sederhana yang siap dibaca pipeline training:

python
# mulai dari kondisi bersih agar run selalu sama
for d in (ORDERS_DIR, FITUR_DIR):
    shutil.rmtree(d, ignore_errors=True)

tulis_partisi(PESANAN, ORDERS_DIR)

# --- Jalankan harian: snapshot 2026-11-01 s.d. 2026-11-05 -----------------------
semua = pd.concat(
    [hitung_fitur(date(2026, 11, d)) for d in range(1, 6)], ignore_index=True
)

# simpan tabel fitur sebagai partisi snapshot_date (offline store sederhana)
for tanggal, grp in semua.groupby("snapshot_date"):
    folder = FITUR_DIR / f"snapshot_date={tanggal}"
    folder.mkdir(parents=True, exist_ok=True)
    pq.write_table(pa.Table.from_pandas(grp), folder / "part.parquet")

Satu pd.concat atas lima pemanggilan hitung_fitur menggantikan lima job terpisah; di produksi, loop ini menjadi jadwal harian di orchestrator dan setiap iterasi menulis satu partisi snapshot.

Output yang Dihasilkan:

text
== Snapshot fitur harian (frek_belanja_7d) ==
customer_id    Budi  Dewi  Siti
snapshot_date                  
2026-11-01      1.0   NaN   1.0
2026-11-02      2.0   1.0   1.0
2026-11-03      2.0   1.0   2.0
2026-11-04      3.0   1.0   2.0
2026-11-05      3.0   2.0   2.0
partisi fitur tersimpan: ['snapshot_date=2026-11-01', 'snapshot_date=2026-11-02', 'snapshot_date=2026-11-03', 'snapshot_date=2026-11-04', 'snapshot_date=2026-11-05']

Nilai NaN pada sel pertama berarti Dewi belum bertransaksi dalam jendela — snapshot memang hanya memuat pelanggan yang aktif pada periode tersebut, dan panjang tabel berubah antar hari. Angka berjalan naik seiring pesanan masuk; pola ini persis cara kerja Incremental Models di section analytics, hanya saja outputnya fitur untuk model, bukan tabel mart untuk dashboard.

Late Event dan Hitung Ulang Snapshot #

Data nyata tidak selalu tiba tepat waktu. Skrip menyimulasikan kasus klasik late arriving data: pesanan Dewi tanggal 2 November baru masuk ke lake pada 6 November, setelah snapshot 4 November dipublikasikan:

python
sebelum = semua[semua["snapshot_date"] == "2026-11-04"].set_index("customer_id")

# --- Late event: pesanan 02 November baru tiba pada 06 November ------------------
LATE = [("ORD-08", "Dewi", datetime(2026, 11, 2, 21, 0), 210_000)]
tulis_partisi(LATE, ORDERS_DIR)

sesudah = hitung_fitur(date(2026, 11, 4)).set_index("customer_id")
banding = sebelum[["frek_belanja_7d", "total_belanja_7d"]].join(
    sesudah[["frek_belanja_7d", "total_belanja_7d"]],
    lsuffix="_sebelum", rsuffix="_sesudah",
).astype("int64")
berubah = banding[banding["frek_belanja_7d_sebelum"] != banding["frek_belanja_7d_sesudah"]]

Karena tulis_partisi menggabungkan alih-alih menimpa, pesanan terlambat masuk tanpa membuang isi partisi lama. Snapshot 4 November lalu dihitung ulang dari sumber yang kini lebih lengkap:

Output yang Dihasilkan:

text
== Late event 02 Nov tiba 06 Nov: snapshot 2026-11-04 dihitung ulang ==
             frek_belanja_7d_sebelum  total_belanja_7d_sebelum  frek_belanja_7d_sesudah  total_belanja_7d_sesudah
customer_id                                                                                                      
Budi                               3                    540000                        3                    540000
Dewi                               1                    350000                        2                    560000
Siti                               2                    120000                        2                    120000
pelanggan yang fiturnya berubah: ['Dewi']

Opsi menyikapi late event	Konsekuensi
Biarkan snapshot lama	fitur basi; model melayani data yang tidak lengkap
Hitung ulang snapshot terdampak (backfill)	benar, tetapi konsumen downstream perlu diberi tahu
Tunda publikasi snapshot beberapa jam	menunggu data tenang, fitur selalu tertinggal

Skrip memilih opsi kedua, dan baris pelanggan yang fiturnya berubah: ['Dewi'] menunjukkan sempitnya dampaknya: hanya pelanggan yang pesanannya datang terlambat perlu diperbarui. Snapshot yang sudah dipakai model lama tidak diubah diam-diam — versinya tercatat di manifest sebagaimana chapter Data Versioning, dan hitung ulang hanya menghasilkan versi baru untuk training berikutnya.

Jadwal harian seperti ini biasanya digantungkan pada orchestrator — materi Scheduling dan Apache Airflow — sehingga backfill snapshot terdampak menjadi satu DAG run ulang, bukan pekerjaan manual yang dikerjakan tengah malam. Namun sebagian use case tidak sanggup menunggu hingga besok: skor fraud dibutuhkan detik itu juga. Kebutuhan itulah yang dibahas chapter Streaming Feature Computation.