Late Arriving Data

Setiap pesanan membawa dua jam yang tidak selalu sepakat: jam ketika transaksi terjadi (event_time) dan jam ketika baris data tiba di pipeline. Ponsel kasir yang offline semalam, antrean aplikasi yang baru terkirim berjam-jam kemudian, atau connector yang di-retry semuanya bisa membuat pesanan kemarin baru muncul hari ini. Untuk laporan penjualan, kebenaran bisnis ada pada event_time — dan itulah mengapa late arriving data mengubah agregat partisi yang mungkin sudah selesai dihitung, bahkan sudah dikirim ke tim keuangan.

Dua Jam yang Berbeda: event_time vs Waktu Kedatangan #

Partisi agregat yang benar mengikuti event_time, sedangkan waktu kedatangan (dimuat_at) hanya menjawab pertanyaan operasional “kapan data ini diproses”. Baris terlambat didefinisikan relatif terhadap partisi, bukan terhadap kalender: sebuah baris dikatakan terlambat bila partisi tanggal event-nya sudah dibangun sebelum baris tersebut tiba. Karena itu kita mencatat dibangun_pada untuk setiap partisi di tabel partisi_state, lalu membandingkannya dengan dimuat_at tiap baris baru.

Dampak ke agregat bisa besar atau kecil tergantung skala hari yang disentuh: satu pesanan Rp150000 menggeser laporan harian toko kecil beberapa persen, tetapi pada hari dengan ratusan juta rupiah pendapatan pergeseran tersebut nyaris tak terlihat — keputusan rebuild tetap diambil demi kebenaran angka, bukan besaran selisihnya.

Tiga Strategi: Rebuild, Jendela Toleransi, dan Koreksi #

Baris terlambat dalam jangka pendek ditangani dengan membangun ulang partisi terdampak dari sumber — murah dan aman karena pola rebuild bersifat idempotent (Chapter Idempotent Transformation). Baris yang terlambat berhari-hari menimbulkan dilema lain: laporan partisi tersebut mungkin sudah dikonsumsi pihak lain, sehingga perubahan angka perlu disetujui dulu. Maka kita memberlakukan jendela toleransi: keterlambatan sampai tiga hari dibangun ulang otomatis; lebih dari itu, partisi menunggu keputusan restatement.

Strategi	Kapan dipakai	Konsekuensi
Rebuild partisi otomatis	keterlambatan dalam jendela toleransi	angka diperbaiki diam-diam, laporan belum terkirim
Tahan + keputusan bisnis	keterlambatan di luar jendela	angka lama dipertahankan sementara
Koreksi (restatement)	keputusan disetujui	laporan resmi diterbitkan ulang

Panjang jendela toleransi adalah kesepakatan, bukan angka teknis. Jendela pendek membuat angka cepat “final” tetapi restatement menjadi sering; jendela panjang mengurangi restatement dengan menunda kepastian. Tim data biasanya menetapkan jendela selaras dengan SLA laporan — bila dashboard harian diterbitkan pukul 07:00 dan konsumen berkomitmen tidak mengutip angka sebelum H+2, jendela dua hari sudah memadai tanpa satu pun restatement.

Implementasi Lengkap: Deteksi, Rebuild, dan Restatement #

Skrip berikut menjalankan tiga babak: tiga batch pagi normal, satu batch malam yang membawa pesanan kemarin, lalu satu baris berumur empat hari yang jatuh di luar toleransi.

python
import sqlite3
from datetime import date
from pathlib import Path

DB = Path("toko.db")
if DB.exists():
    DB.unlink()

koneksi = sqlite3.connect(DB)
koneksi.executescript("""
CREATE TABLE pesanan (
    id INTEGER PRIMARY KEY,
    event_time TEXT NOT NULL,
    dimuat_at TEXT NOT NULL,
    pelanggan TEXT NOT NULL,
    qty INTEGER NOT NULL,
    harga INTEGER NOT NULL
);
INSERT INTO pesanan (event_time, dimuat_at, pelanggan, qty, harga) VALUES
    ('2026-01-09T10:00:00', '2026-01-10T06:00:00', 'Budi', 1, 250000),
    ('2026-01-09T15:30:00', '2026-01-10T06:00:00', 'Siti', 2, 180000),
    ('2026-01-10T09:15:00', '2026-01-11T06:00:00', 'Dewi', 1, 120000),
    ('2026-01-10T14:30:00', '2026-01-11T06:00:00', 'Budi', 1, 210000),
    ('2026-01-10T19:45:00', '2026-01-11T06:00:00', 'Rina', 3, 250000),
    ('2026-01-11T08:05:00', '2026-01-12T06:00:00', 'Siti', 2, 150000),
    ('2026-01-11T13:20:00', '2026-01-12T06:00:00', 'Dewi', 1, 180000);
CREATE TABLE penjualan_harian (
    tanggal TEXT PRIMARY KEY,
    jumlah_pesanan INTEGER NOT NULL,
    total_pendapatan INTEGER NOT NULL
);
CREATE TABLE partisi_state (
    tanggal TEXT PRIMARY KEY,
    dibangun_pada TEXT NOT NULL
);
""")

TOLERANSI_HARI = 3


def bangun_partisi(tanggal_event, waktu_proses):
    koneksi.execute("DELETE FROM penjualan_harian WHERE tanggal = ?", (tanggal_event,))
    koneksi.execute(
        """INSERT INTO penjualan_harian (tanggal, jumlah_pesanan, total_pendapatan)
           SELECT substr(event_time, 1, 10), COUNT(*), SUM(qty * harga)
           FROM pesanan
           WHERE substr(event_time, 1, 10) = ?
           GROUP BY substr(event_time, 1, 10)""",
        (tanggal_event,),
    )
    koneksi.execute("INSERT OR REPLACE INTO partisi_state VALUES (?, ?)",
                    (tanggal_event, waktu_proses))
    koneksi.commit()


def tampil(judul):
    print(f"=== {judul} ===")
    for tanggal, jumlah, total in koneksi.execute(
        "SELECT tanggal, jumlah_pesanan, total_pendapatan FROM penjualan_harian ORDER BY tanggal"
    ):
        print(f"  {tanggal} | {jumlah} pesanan | Rp{total}")


def proses_batch(waktu_proses):
    (build_terakhir,) = koneksi.execute(
        "SELECT COALESCE(MAX(dibangun_pada), '0000-01-01T00:00:00') FROM partisi_state"
    ).fetchone()
    baris_baru = koneksi.execute(
        """SELECT id, event_time, dimuat_at FROM pesanan
           WHERE dimuat_at > ? AND dimuat_at <= ? ORDER BY id""",
        (build_terakhir, waktu_proses),
    ).fetchall()

    state = dict(koneksi.execute("SELECT tanggal, dibangun_pada FROM partisi_state"))
    perlu_rebuild, perlu_manual = set(), set()

    for id_, event_time, dimuat_at in baris_baru:
        tanggal_event = event_time[:10]
        dibangun_pada = state.get(tanggal_event)
        if dibangun_pada is None:
            print(f"  pesanan {id_}: event {tanggal_event} -> partisi baru")
            perlu_rebuild.add(tanggal_event)
        elif dibangun_pada < dimuat_at:
            selisih = (date.fromisoformat(waktu_proses[:10])
                       - date.fromisoformat(tanggal_event)).days
            if selisih <= TOLERANSI_HARI:
                print(f"  pesanan {id_}: event {tanggal_event} -> TERLAMBAT {selisih} hari, rebuild partisi")
                perlu_rebuild.add(tanggal_event)
            else:
                print(f"  pesanan {id_}: event {tanggal_event} -> TERLAMBAT {selisih} hari, DI LUAR toleransi")
                perlu_manual.add(tanggal_event)

    for tanggal_event in sorted(perlu_rebuild - perlu_manual):
        bangun_partisi(tanggal_event, waktu_proses)
    for tanggal_event in sorted(perlu_manual):
        print(f"  partisi {tanggal_event} menunggu keputusan bisnis (laporan sudah terkirim)")


# Riwayat: tiga batch pagi yang berjalan normal
for tanggal, waktu in [("2026-01-09", "2026-01-10T06:30:00"),
                       ("2026-01-10", "2026-01-11T06:30:00"),
                       ("2026-01-11", "2026-01-12T06:30:00")]:
    bangun_partisi(tanggal, waktu)
tampil("SEBELUM: agregat tiga batch normal")

# Sore 12 Januari: pesanan kemarin luput dari batch pagi
koneksi.executemany(
    "INSERT INTO pesanan (event_time, dimuat_at, pelanggan, qty, harga) VALUES (?, ?, ?, ?, ?)",
    [
        ("2026-01-11T21:10:00", "2026-01-12T20:55:00", "Rina", 1, 300000),
        ("2026-01-12T07:40:00", "2026-01-12T20:55:00", "Budi", 1, 180000),
    ],
)
koneksi.commit()
print("=== Batch malam 2026-01-12 ===")
proses_batch("2026-01-12T21:05:00")
tampil("SESUDAH: partisi 2026-01-11 dibangun ulang")

# Pagi 13 Januari: pesanan berumur 4 hari baru tiba
koneksi.execute(
    "INSERT INTO pesanan (event_time, dimuat_at, pelanggan, qty, harga) VALUES (?, ?, ?, ?, ?)",
    ("2026-01-09T21:30:00", "2026-01-13T06:00:00", "Siti", 1, 150000),
)
koneksi.commit()
print("=== Batch pagi 2026-01-13 ===")
proses_batch("2026-01-13T06:30:00")

print("=== Koreksi manual: restatement partisi 2026-01-09 ===")
bangun_partisi("2026-01-09", "2026-01-13T09:00:00")
tampil("FINAL: setelah restatement")

koneksi.close()

Output yang Dihasilkan:

text
=== SEBELUM: agregat tiga batch normal ===
  2026-01-09 | 2 pesanan | Rp610000
  2026-01-10 | 3 pesanan | Rp1080000
  2026-01-11 | 2 pesanan | Rp480000
=== Batch malam 2026-01-12 ===
  pesanan 8: event 2026-01-11 -> TERLAMBAT 1 hari, rebuild partisi
  pesanan 9: event 2026-01-12 -> partisi baru
=== SESUDAH: partisi 2026-01-11 dibangun ulang ===
  2026-01-09 | 2 pesanan | Rp610000
  2026-01-10 | 3 pesanan | Rp1080000
  2026-01-11 | 3 pesanan | Rp780000
  2026-01-12 | 1 pesanan | Rp180000
=== Batch pagi 2026-01-13 ===
  pesanan 10: event 2026-01-09 -> TERLAMBAT 4 hari, DI LUAR toleransi
  partisi 2026-01-09 menunggu keputusan bisnis (laporan sudah terkirim)
=== Koreksi manual: restatement partisi 2026-01-09 ===
=== FINAL: setelah restatement ===
  2026-01-09 | 3 pesanan | Rp760000
  2026-01-10 | 3 pesanan | Rp1080000
  2026-01-11 | 3 pesanan | Rp780000
  2026-01-12 | 1 pesanan | Rp180000

Mari kita telaah baris-baris kode di atas:

Baris 41-53 (bangun_partisi): partisi tanggal event dihapus lalu diisi ulang dari seluruh baris sumber partisi tersebut, dan partisi_state mencatat waktu_proses sebagai waktu pembangunan — pola rebuild idempotent dari chapter sebelumnya.
Baris 64-72 (proses_batch, pengambilan delta): batch memproses baris dengan dimuat_at setelah build terakhir dan sebelum waktu_proses; batas atas menjaga baris “dari masa depan” agar tidak ikut diproses.
Baris 77-91 (klasifikasi tiap baris): baris dengan partisi yang belum ada menjadi partisi baru; baris yang tiba setelah partisinya dibangun dihitung selisih harinya, lalu diarahkan ke rebuild otomatis atau antrean keputusan manual.
Baris 93-96 (eksekusi keputusan): himpunan perlu_rebuild - perlu_manual memastikan partisi yang butuh keputusan bisnis tidak ikut dibangun ulang diam-diam.

Pesanan 8 (event 11 Januari, tiba malam 12 Januari) memicu rebuild partisi 2026-01-11: agregat naik dari 2 menjadi 3 pesanan dan Rp480000 menjadi Rp780000, sementara pesanan 9 yang tepat waktu membuka partisi baru 2026-01-12. Pesanan 10 yang terlambat empat hari ditahan; setelah keputusan restatement disetujui, partisi 2026-01-09 diperbaiki menjadi 3 pesanan Rp760000.

Rebuild dan restatement memperbaiki angka, tetapi bagaimana kita tahu tabel agregat benar-benar cocok dengan sumbernya setelah semua peristiwa tersebut? Chapter Data Reconciliation membangun pemeriksaan otomatis yang membandingkan sumber dan target baris demi baris maupun lewat checksum agregat.