1 Identitas dan Judul Dokumen

Keterangan Isi
Tautan Video https://youtu.be/jWwTEHa5Xis
Mata kuliah Pengantar Big Data
Nama Alya Maura Raditha
NIM 52240003
Program studi Sains Data, Institut Teknologi Sains Bandung (ITSB)
Topik Studi kasus aplikasi dompet digital BayarYuk
Teknologi Python, PySpark (Apache Spark), simulasi MongoDB (mongomock), R Markdown

2 Pendahuluan

Big Data tidak hanya berarti data yang besar. Istilah ini merujuk pada data yang volumenya, kecepatan datangnya, dan ragam bentuknya sudah melampaui kemampuan satu mesin atau satu basis data relasional. Karena itu satu analisis sering harus menggabungkan beberapa teknologi, dan dokumen ini memperlihatkan hal tersebut dalam satu alur kerja:

  1. Simulasi MongoDB (mongomock) menyimpan data merchant yang strukturnya fleksibel (schema-flexible).
  2. Apache Spark (RDD) mengolah log transaksi berukuran 400.000 baris yang sebagian barisnya rusak.
  3. Spark DataFrame dan Spark SQL mengintegrasikan data merchant dengan transaksi dan menghasilkan agregasi.
  4. Spark Structured Streaming meng-ingest transaksi baru dari landing zone secara bertahap dan menyimpannya ke format Parquet.
  5. Exploratory Data Analysis (EDA) pada data skala besar menghasilkan insight untuk keputusan bisnis.

Dokumen ini disusun dalam urutan pengerjaan: konsep dasar, lalu implementasi C1 sampai C5. Setiap bagian implementasi memuat tujuan, implementasi, serta hasil dan interpretasi. Seluruh angka pada dokumen ini keluar dari eksekusi kode saat dokumen di-knit, bukan diketik manual.

3 Deskripsi Studi Kasus BayarYuk

BayarYuk adalah aplikasi dompet digital dengan beragam aktivitas transaksi. Tim data BayarYuk diminta menganalisis tiga sumber data yang karakteristiknya berbeda, yaitu data merchant, log transaksi, dan aliran transaksi baru (streaming). Tujuannya:

  • mengolah data dari berbagai sumber;
  • menangani masalah kualitas data (nilai sistem error, nilai ekstrem, dan baris rusak);
  • mengagregasi transaksi dan mengintegrasikannya dengan data merchant;
  • membangun proses ingestion yang tahan gagal (fault tolerant);
  • menghasilkan insight yang mendukung keputusan bisnis.

4 Deskripsi Sumber Data dan Karakteristik Dataset

4.1 Ringkasan Sumber Data

Sumber Wadah Karakteristik Dipakai pada
A. Merchant MongoDB (simulator mongomock), db["merchant"] 200 dokumen; _id, nama, kategori (5 jenis), kota (5 kota), rating, promo (array 0–2 elemen); sekitar 30% memiliki jam_buka C1, C3
B. Log transaksi File teks, variabel jalur_log 400.000 baris, pemisah \|: waktu\|id_trx\|id_user\|id_merchant\|nominal\|status; ada nominal -1, nominal Rp250.000.000, dan 80 baris rusak (baris 1, 5.001, 10.001, dan seterusnya) C2, C3, C5
C. Landing zone Folder berisi 3 file JSON 3 file × 500 event = 1.500 event; id_trx, id_merchant, nominal, waktu; sekitar 3% nominal bernilai null C4

4.2 Persiapan Lingkungan

Bagian ini menyiapkan Python, Spark, dan simulator MongoDB. Konfigurasi mengikuti lingkungan Windows yang dipakai pada pengerjaan ini.

4.2.1 Konfigurasi yang Perlu Disesuaikan

Konfigurasi Nilai yang dipakai Cara menyesuaikan
Python Python 3.11 pada py_path (chunk setup) ubah py_path ke lokasi python.exe
Java D:/Java ubah JAVA_HOME pada chunk setup dan lingkungan-spark
Hadoop D:/hadoop, berisi bin/winutils.exe dan hadoop.dll ubah HADOOP_HOME pada chunk setup dan lingkungan-spark
Folder sementara Spark D:/spark-temp ubah SPARK_LOCAL_DIR pada chunk lingkungan-spark
Folder keluaran data_bayaryuk/ di samping file .Rmd (Parquet, checkpoint, log Spark) set variabel lingkungan BAYARYUK_DATA_DIR
Paket Python pyspark, mongomock, numpy, pandas, matplotlib pip install -r requirements.txt
Paket R rmarkdown, knitr, reticulate, plotly install.packages(c("rmarkdown", "knitr", "reticulate", "plotly"))

4.2.2 Konfigurasi Python dan Impor Library

4.2.3 Membuat SparkSession

Chunk setup-spark di bawah ini disalin tanpa perubahan dari file acuan UTS. Pengaturan khusus Windows (Java, Hadoop, folder sementara) ada pada chunk lingkungan-spark di atasnya.

# Mengimpor modul sistem untuk mengatur lingkungan Python.
import os, sys

# Driver dan worker PySpark memakai interpreter Python yang sama.
os.environ.setdefault("PYSPARK_PYTHON", sys.executable)
## 'C:\\Users\\Dell\\AppData\\Local\\Programs\\Python\\PYTHON~3\\python.exe'
os.environ.setdefault("PYSPARK_DRIVER_PYTHON", sys.executable)
## 'C:\\Users\\Dell\\AppData\\Local\\Programs\\Python\\PYTHON~3\\python.exe'
# Mengimpor SparkSession sebagai pintu masuk fitur Spark.
from pyspark.sql import SparkSession

# Membuat sesi Spark mode lokal untuk praktikum.
spark = (
    SparkSession.builder

    # Memakai seluruh core CPU laptop.
    .master("local[*]")

    # Nama aplikasi Spark.
    .appName("BigData-P08")

    # Mematikan bar progres agar output dokumen rapi.
    .config("spark.ui.showConsoleProgress", "false")

    # Partisi shuffle 8 (default 200 terlalu besar untuk praktikum).
    .config("spark.sql.shuffle.partitions", "8")

    # Memori driver 2 GB.
    .config("spark.driver.memory", "2g")

    # Membuat sesi baru atau memakai sesi yang sudah ada.
    .getOrCreate()
)

# Hanya menampilkan log tingkat ERROR.
spark.sparkContext.setLogLevel("ERROR")

# Menyimpan referensi SparkContext.
sc = spark.sparkContext

# Menampilkan versi, mode master, dan jumlah core paralel.
print(
    "Spark versi", spark.version,
    "| master:", sc.master,
    "| jumlah core paralel:", sc.defaultParallelism
)
## Spark versi 3.5.9 | master: local[*] | jumlah core paralel: 8

Zona waktu sesi Spark disamakan dengan waktu pada log (UTC), agar fungsi hour() pada C5 membaca jam apa adanya dan tidak bergeser mengikuti zona waktu komputer.

spark.conf.set("spark.sql.session.timeZone", "UTC")

4.2.4 Pemeriksaan Hadoop di Windows

Penulisan Parquet di Windows membutuhkan winutils.exe dan hadoop.dll. Pemeriksaan berikut memastikan keduanya tersedia dan versi Hadoop yang dibawa Spark terbaca.

4.3 Pembangkitan Dataset

Dataset dibangkitkan oleh chunk uts-data, yang disalin tanpa perubahan dari file acuan UTS (BD_P08_UTS.Rmd): seed 2026, urutan pemanggilan generator acak, dan parameter distribusinya sama, sehingga datanya identik dengan dataset acuan dan tetap sama setiap kali dokumen dijalankan ulang. Data merchant disimpan pada simulator MongoDB mongomock, sehingga tidak diperlukan server MongoDB. Dataset ini bersifat simulasi: masalah yang disuntikkan secara sengaja hanyalah nominal -1, nominal Rp250.000.000, baris log rusak, dan nominal null pada data streaming. Waktu transaksi, merchant, dan status tidak diberi pola khusus, sehingga kesimpulan analisis berlaku untuk data simulasi ini saja.

# Mengimpor pustaka yang dipakai selama ujian.
import numpy as np
import pandas as pd
import os
import json
import tempfile
import mongomock

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# Seed tetap agar semua peserta mendapat data yang sama.
rng = np.random.default_rng(2026)
folder = tempfile.mkdtemp()

# ============================================================
# 1. DATA MERCHANT (SIMULASI BASIS DATA MONGODB)
# ============================================================

kategori = ["Kuliner", "Retail", "Transportasi", "Hiburan", "Kesehatan"]
kota = ["Jakarta", "Bandung", "Surabaya", "Medan", "Makassar"]

merchant = []

for i in range(1, 201):

    # Setiap merchant memiliki 0-2 promo acak.
    d = {
        "_id": f"M{i:03d}",
        "nama": f"Merchant {i}",
        "kategori": str(rng.choice(kategori)),
        "kota": str(rng.choice(kota)),
        "rating": float(np.round(rng.uniform(3.0, 5.0), 1)),
        "promo": [
            str(p) for p in rng.choice(
                ["cashback", "diskon", "gratis_ongkir", "poin"],
                size=rng.integers(0, 3),
                replace=False
            )
        ]
    }

    # 30% merchant memiliki field jam_buka (skema fleksibel).
    if rng.random() < 0.3:
        d["jam_buka"] = {
            "buka": "08:00",
            "tutup": str(rng.choice(["21:00", "22:00", "23:59"]))
        }

    merchant.append(d)

# Menyimpan merchant ke simulasi basis data MongoDB menggunakan mongomock.
db = mongomock.MongoClient()["bayaryuk"]
db["merchant"].insert_many(merchant)
## InsertManyResult(['M001', 'M002', 'M003', 'M004', 'M005', 'M006', 'M007', 'M008', 'M009', 'M010', 'M011', 'M012', 'M013', 'M014', 'M015', 'M016', 'M017', 'M018', 'M019', 'M020', 'M021', 'M022', 'M023', 'M024', 'M025', 'M026', 'M027', 'M028', 'M029', 'M030', 'M031', 'M032', 'M033', 'M034', 'M035', 'M036', 'M037', 'M038', 'M039', 'M040', 'M041', 'M042', 'M043', 'M044', 'M045', 'M046', 'M047', 'M048', 'M049', 'M050', 'M051', 'M052', 'M053', 'M054', 'M055', 'M056', 'M057', 'M058', 'M059', 'M060', 'M061', 'M062', 'M063', 'M064', 'M065', 'M066', 'M067', 'M068', 'M069', 'M070', 'M071', 'M072', 'M073', 'M074', 'M075', 'M076', 'M077', 'M078', 'M079', 'M080', 'M081', 'M082', 'M083', 'M084', 'M085', 'M086', 'M087', 'M088', 'M089', 'M090', 'M091', 'M092', 'M093', 'M094', 'M095', 'M096', 'M097', 'M098', 'M099', 'M100', 'M101', 'M102', 'M103', 'M104', 'M105', 'M106', 'M107', 'M108', 'M109', 'M110', 'M111', 'M112', 'M113', 'M114', 'M115', 'M116', 'M117', 'M118', 'M119', 'M120', 'M121', 'M122', 'M123', 'M124', 'M125', 'M126', 'M127', 'M128', 'M129', 'M130', 'M131', 'M132', 'M133', 'M134', 'M135', 'M136', 'M137', 'M138', 'M139', 'M140', 'M141', 'M142', 'M143', 'M144', 'M145', 'M146', 'M147', 'M148', 'M149', 'M150', 'M151', 'M152', 'M153', 'M154', 'M155', 'M156', 'M157', 'M158', 'M159', 'M160', 'M161', 'M162', 'M163', 'M164', 'M165', 'M166', 'M167', 'M168', 'M169', 'M170', 'M171', 'M172', 'M173', 'M174', 'M175', 'M176', 'M177', 'M178', 'M179', 'M180', 'M181', 'M182', 'M183', 'M184', 'M185', 'M186', 'M187', 'M188', 'M189', 'M190', 'M191', 'M192', 'M193', 'M194', 'M195', 'M196', 'M197', 'M198', 'M199', 'M200'], acknowledged=True)
# ============================================================
# 2. LOG TRANSAKSI (400 RIBU BARIS)
# Format: waktu|id_trx|id_user|id_merchant|nominal|status
# ============================================================

n = 400_000

waktu = (
    pd.Timestamp("2026-09-01")
    + pd.to_timedelta(rng.integers(0, 30 * 86400, n), unit="s")
)

id_m = rng.choice([m["_id"] for m in merchant], n)

nominal = np.round(
    rng.lognormal(10.6, 0.9, n), -2
).astype(int)

status = rng.choice(
    ["SUKSES", "GAGAL", "PENDING"],
    n,
    p=[0.93, 0.05, 0.02]
)

# Menyuntikkan kesalahan sistem (-1) dan nilai ekstrem mencurigakan.
nominal[rng.choice(n, 150, replace=False)] = -1
nominal[rng.choice(n, 80, replace=False)] = 250_000_000

# Menyusun baris log berpemisah "|".
baris = [
    f"{w:%Y-%m-%d %H:%M:%S}|T{i:07d}|U{u:05d}|{m}|{v}|{s}"
    for i, (w, u, m, v, s) in enumerate(
        zip(
            waktu,
            rng.integers(1, 50_000, n),
            id_m,
            nominal,
            status
        )
    )
]

# Setiap baris ke-5.000 diganti dengan baris log rusak.
baris[::5000] = [
    "ERROR koneksi timeout"
] * len(baris[::5000])

# Menyimpan log transaksi ke berkas teks.
jalur_log = os.path.join(folder, "transaksi.log")

with open(jalur_log, "w", encoding="utf-8") as f:
    _ = f.write("\n".join(baris) + "\n")


# ============================================================
# 3. LANDING ZONE UNTUK STREAMING
# 3 berkas JSON, masing-masing 500 data
# ============================================================

landing = os.path.join(folder, "landing")
os.makedirs(landing, exist_ok=True)

for b in range(3):

    jalur_json = os.path.join(
        landing,
        f"batch_{b}.json"
    )

    with open(jalur_json, "w", encoding="utf-8") as f:

        for k in range(500):

            e = {
                "id_trx": f"S{b}{k:04d}",
                "id_merchant": str(rng.choice(id_m)),
                "nominal": (
                    int(rng.lognormal(10.6, 0.9))
                    if rng.random() > 0.03
                    else None
                ),
                "waktu": str(
                    pd.Timestamp("2026-10-01 09:00")
                    + pd.Timedelta(minutes=b * 5 + k % 5)
                )
            }

            # Menulis satu data JSON per baris tanpa menampilkan
            # nilai kembalian dari operasi penulisan.
            _ = f.write(json.dumps(e) + "\n")


# ============================================================
# 4. RINGKASAN DATASET UJIAN
# ============================================================

print(
    "Merchant:", db["merchant"].count_documents({}),
    "| baris log:", len(baris),
    "| berkas landing:", len(os.listdir(landing))
)
## Merchant: 200 | baris log: 400000 | berkas landing: 3

Chunk uts-data menyimpan log dan folder JSON pada folder sementara (jalur_log dan landing). Nama folder_landing yang dipakai pada C4 disamakan dengan landing, dan jalur keduanya diseragamkan ke format /.

# jalur diseragamkan ke format "/" agar terbaca Spark di Windows
jalur_log = Path(jalur_log).as_posix()
folder_landing = Path(landing).as_posix()

4.4 Verifikasi Dataset

Sebelum analisis, setiap karakteristik pada spesifikasi diperiksa ulang secara independen: log dan JSON dibaca dengan Python murni (bukan Spark) dan data merchant dihitung dengan query-nya sendiri.

# Verifikasi A: merchant pada simulasi MongoDB (mongomock)
col = db["merchant"]
print("== A. MERCHANT (simulasi MongoDB, mongomock) ==")
## == A. MERCHANT (simulasi MongoDB, mongomock) ==
print("Jumlah dokumen            :", col.count_documents({}))
## Jumlah dokumen            : 200
print("Kategori distinct         :", len(col.distinct("kategori")), sorted(col.distinct("kategori")))
## Kategori distinct         : 5 ['Hiburan', 'Kesehatan', 'Kuliner', 'Retail', 'Transportasi']
print("Kota distinct             :", len(col.distinct("kota")), sorted(col.distinct("kota")))
## Kota distinct             : 5 ['Bandung', 'Jakarta', 'Makassar', 'Medan', 'Surabaya']
n_jam = col.count_documents({"jam_buka": {"$exists": True}})
print(f"Memiliki jam_buka         : {n_jam} ({n_jam/200:.0%})")
## Memiliki jam_buka         : 61 (30%)
print("Panjang array promo       :", {r["_id"]: r["n"] for r in col.aggregate(
    [{"$group": {"_id": {"$size": "$promo"}, "n": {"$sum": 1}}}, {"$sort": {"_id": 1}}])})
## Panjang array promo       : {0: 67, 1: 70, 2: 63}
# Verifikasi B: log transaksi, dibaca baris demi baris dengan Python murni (tidak dimuat seluruhnya ke memori)
from collections import Counter
n_baris, rusak, contoh_awal, contoh_rusak = 0, [], [], []
n_negatif = n_ekstrem = 0
hitung_status = Counter()
with open(jalur_log, encoding="utf-8") as fh:
    for i, b in enumerate(fh, start=1):
        b = b.rstrip("\n")
        n_baris += 1
        if i <= 3:
            contoh_awal.append(b)
        f = b.split("|")
        try:
            if len(f) != 6:
                raise ValueError
            nom_i = int(f[4])
        except ValueError:
            rusak.append(i)
            if len(contoh_rusak) < 2:
                contoh_rusak.append(b)
            continue
        n_negatif += nom_i == -1
        n_ekstrem += nom_i == 250_000_000
        hitung_status[f[5]] += 1

print("\n== B. LOG TRANSAKSI ==")
## 
## == B. LOG TRANSAKSI ==
print("Jumlah baris              :", n_baris)
## Jumlah baris              : 400000
print("3 baris pertama           :")
## 3 baris pertama           :
for b in contoh_awal:
    print("   ", b)
##     ERROR koneksi timeout
##     2026-09-20 01:13:10|T0000001|U42902|M172|25000|SUKSES
##     2026-09-29 10:06:45|T0000002|U03879|M041|196700|SUKSES
print("Baris rusak               :", len(rusak), "| posisi baris 1, 5.001, 10.001, ... :", all((i - 1) % 5000 == 0 for i in rusak))
## Baris rusak               : 80 | posisi baris 1, 5.001, 10.001, ... : True
print("Contoh baris rusak        :", contoh_rusak)
## Contoh baris rusak        : ['ERROR koneksi timeout', 'ERROR koneksi timeout']
print("Nominal = -1              :", n_negatif)
## Nominal = -1              : 150
print("Nominal = 250.000.000     :", n_ekstrem)
## Nominal = 250.000.000     : 80
print("Distribusi status         :", dict(hitung_status))
## Distribusi status         : {'SUKSES': 371870, 'GAGAL': 20166, 'PENDING': 7884}
# Verifikasi C: landing zone JSON
print("\n== C. LANDING ZONE ==")
## 
## == C. LANDING ZONE ==
total, null_total = 0, 0
for p in sorted(Path(folder_landing).glob("*.json")):
    ev = [json.loads(l) for l in p.read_text().splitlines()]
    k = sum(e["nominal"] is None for e in ev); total += len(ev); null_total += k
    print(f"{p.name}: {len(ev)} event, nominal null = {k}")
## batch_0.json: 500 event, nominal null = 14
## batch_1.json: 500 event, nominal null = 21
## batch_2.json: 500 event, nominal null = 18
print(f"Total event: {total} | nominal null: {null_total} ({null_total/total:.1%})")
## Total event: 1500 | nominal null: 53 (3.5%)

5 Jawaban Pertanyaan Konseptual

5.1 Bagian A – Pilihan Ganda

A1. Jawaban: b. Velocity. Alasan: Dua juta pengemudi yang mengirim posisi setiap 3 detik menghasilkan sekitar 667 ribu event per detik, sehingga ciri yang paling menonjol adalah laju kedatangan data, bukan ragam format (seluruhnya data GPS seragam), keandalan, atau nilai.

A2. Jawaban: c. Redis. Alasan: Sesi login adalah pasangan kunci–nilai berumur pendek yang sering dibaca dan ditulis dengan latensi sangat rendah, dan Redis menyediakan kedaluwarsa otomatis (TTL) bawaan pada setiap kunci.

A3. Jawaban: c. $unwind. Alasan: $unwind memecah field array sehingga setiap elemen menjadi dokumen tersendiri, sedangkan $match menyaring, $group mengelompokkan, dan $project memilih atau membentuk field.

A4. Jawaban: b. 6 block, 2.100 MB. Alasan: 700 MB ÷ 128 MB = 5,47 sehingga dibutuhkan 6 block (lima block penuh 640 MB dan satu block terakhir 60 MB). Dengan replikasi 3, total data tersimpan adalah 700 × 3 = 2.100 MB, dan jumlah replika block menjadi 6 × 3 = 18.

A5. Jawaban: b. NameNode. Alasan: NameNode menyimpan metadata sistem berkas (struktur direktori, izin, dan pemetaan berkas ke block) serta mengetahui DataNode mana yang memegang tiap block dari laporan block yang dikirim DataNode.

A6. Jawaban: d. count(). Alasan: count() memaksa Spark menjalankan seluruh rencana komputasi dan mengembalikan hasil ke driver, sedangkan map(), filter(), dan reduceByKey() hanya mendefinisikan transformasi baru dan tidak langsung dieksekusi.

A7. Jawaban: b. reduceByKey(). Alasan: reduceByKey menjumlahkan nilai di tiap partisi lebih dulu sebelum shuffle (map-side combine), sehingga data yang dikirim antar node jauh lebih sedikit dibanding groupByKey yang mengirim semua nilai.

A8. Jawaban: b. Alasan: Explicit schema menghindari pemindaian tambahan yang dibutuhkan inferSchema untuk menebak tipe, mencegah salah tebak (misalnya ID 007 terbaca sebagai angka, atau satu nilai kotor mengubah seluruh kolom menjadi string), dan berfungsi sebagai kontrak yang terdokumentasi.

A9. Jawaban: b. 4. Alasan: Dalam satu consumer group, setiap partisi hanya boleh dibaca oleh satu consumer, sehingga jumlah consumer aktif paling banyak sama dengan jumlah partisi, yaitu 4. Dua consumer sisanya menganggur (siaga bila ada consumer lain gagal).

A10. Jawaban: c. Alasan: Dua miliar baris tidak muat di memori driver, sehingga toPandas() dan collect() akan gagal, sedangkan take(10) hanya memberi sampel kecil yang tidak mewakili distribusi. Pendekatan yang benar adalah membagi nilai ke dalam interval (bin) dan menghitungnya di Spark, lalu memvisualisasikan hasil agregasi yang kecil.

5.2 Bagian B – Esai

5.2.1 B1. Scale-up vs Scale-out dan Mahalnya Shuffle

Perbedaan. Scale-up (vertikal) menambah kekuatan satu mesin, misalnya CPU, RAM, dan disk yang lebih besar. Scale-out (horizontal) menambah jumlah mesin biasa yang bekerja bersama dalam satu klaster.

Aspek Scale-up Scale-out
Batas pertumbuhan Dibatasi spesifikasi tertinggi satu mesin Praktis dapat terus ditambah node
Biaya Naik tajam (tidak linear) pada spesifikasi tinggi Cenderung linear dengan perangkat standar
Ketahanan Satu mesin = satu titik kegagalan (single point of failure) Kegagalan satu node dapat diatasi node lain
Kompleksitas perangkat lunak Rendah, aplikasi biasa tetap berjalan Tinggi: data harus dipartisi dan kegagalan harus ditangani
Contoh Meningkatkan server PostgreSQL tunggal Klaster HDFS dan Spark, MongoDB sharded

Big Data memilih scale-out karena volume data melampaui kemampuan satu mesin dan toleransi kegagalan diperlukan. Harga yang dibayar adalah kompleksitas, yang diselesaikan oleh HDFS (penyimpanan terdistribusi) serta MapReduce dan Spark (komputasi terdistribusi).

Mengapa shuffle mahal. Shuffle adalah pengelompokan ulang data antar partisi dan antar node berdasarkan key, yang diperlukan oleh operasi seperti groupByKey, reduceByKey, join, dan repartition. Biayanya berlapis:

  1. Jaringan. Setiap tugas map dapat mengirim data ke banyak tugas reduce (pola semua-ke-semua), dan jaringan jauh lebih lambat daripada memori.
  2. Disk. Keluaran map ditulis ke disk lokal (pada MapReduce dan juga pada berkas shuffle Spark) agar dapat diambil reducer dan aman bila terjadi kegagalan.
  3. Serialisasi dan pengurutan/penggabungan. Data harus diubah menjadi byte dan dikembalikan, lalu diurutkan atau digabung per key, yang memakan CPU dan memori (dapat spill ke disk).
  4. Sinkronisasi. Tahap berikutnya baru dapat mulai setelah seluruh tahap sebelumnya selesai, sehingga satu tugas lambat (misalnya karena data skew: satu key sangat dominan) menahan seluruh pekerjaan.

Karena itu shuffle diminimalkan, misalnya dengan reduceByKey dan broadcast join (C2 dan C3).

5.2.2 B2. Pilihan Database NoSQL untuk Marketplace

Kebutuhan Jenis NoSQL Contoh
1. Katalog produk dengan atribut berbeda per kategori Document store MongoDB
2. Keranjang belanja sementara Key-value store Redis
3. Rekomendasi “friends also bought” Graph database Neo4j
  1. Katalog produk: document store. Laptop memiliki atribut seperti RAM dan prosesor, kaos memiliki ukuran dan warna, dan buku memiliki penulis dan ISBN. Dokumen JSON fleksibel menyimpan atribut berbeda dalam koleksi yang sama tanpa kolom NULL atau tabel yang terus diubah. Array dan objek bersarang (misalnya daftar varian) dapat disimpan dalam satu dokumen, dan field dapat diindeks untuk pencarian. Fleksibilitas yang sama terlihat pada field jam_buka merchant BayarYuk.
  2. Keranjang belanja: key-value store. Akses selalu berdasarkan satu kunci (ID pengguna atau sesi), isinya sementara, dan harus cepat dibaca serta ditulis. Redis menyimpan data di memori dengan latensi sangat rendah dan TTL otomatis untuk menghapus keranjang yang ditinggalkan. Jika keranjang tidak boleh hilang saat restart, opsi persistensi (RDB atau AOF) dapat diaktifkan.
  3. Rekomendasi “friends also bought”: graph database. Kueri ini menelusuri relasi berantai: pengguna → teman → produk yang dibeli teman. Pada graph database, penelusuran relasi adalah operasi utama dan relasi disimpan langsung sebagai penghubung antar simpul, sehingga biaya penelusuran tidak membengkak seiring kedalaman. Pada basis data relasional, kueri yang sama membutuhkan banyak JOIN bertingkat yang makin lambat pada data besar.

5.2.3 B3. Lazy Evaluation dan Lineage di Apache Spark

Lazy evaluation. Transformation seperti map, filter, dan join tidak langsung dijalankan. Spark hanya mencatatnya sebagai rencana (DAG), dan baru mengeksekusinya ketika ada action seperti count, collect, atau write. Pada C2, baris rdd_mentah.flatMap(parse_baris) belum membaca file sama sekali sampai count() dipanggil.

Lineage. Setiap RDD atau DataFrame mengingat dari mana ia berasal: RDD induk dan transformation yang membentuknya. Rantai riwayat ini disebut lineage, dan merupakan bentuk konkret dari DAG yang dibangun oleh lazy evaluation.

Manfaat dalam pengolahan terdistribusi:

  1. Optimasi. Karena seluruh rencana diketahui sebelum dijalankan, Spark dapat menggabungkan transformation sempit (map, filter) ke dalam satu tahap tanpa menyimpan hasil antara, menentukan batas shuffle, dan (pada DataFrame lewat Catalyst) menerapkan predicate pushdown dan column pruning.
  2. Menghemat komputasi dan memori. Pekerjaan yang tidak dibutuhkan tidak dikerjakan; misalnya take(5) hanya menghitung partisi yang diperlukan.
  3. Toleransi kegagalan tanpa replikasi data. Bila satu node gagal dan sebuah partisi hilang, Spark menghitung ulang hanya partisi itu dari sumber menurut lineage, tanpa menyimpan salinan setiap hasil antara. Ini jauh lebih murah daripada replikasi data perantara. Untuk lineage yang sangat panjang atau hasil yang dipakai berulang, cache() atau checkpoint dipakai agar pemulihan tidak mengulang dari awal.

5.2.4 B4. At-most-once vs At-least-once dan Peran Checkpoint

Aspek At-most-once At-least-once
Jaminan Pesan diproses 0 atau 1 kali Pesan diproses 1 kali atau lebih
Risiko Data hilang, tanpa duplikat Duplikat, tanpa kehilangan
Mekanisme Tanpa percobaan ulang; offset dicatat sebelum diproses Dicoba ulang sampai berhasil; offset dicatat setelah diproses
Contoh kegagalan Crash setelah offset dicatat tetapi sebelum diproses: pesan tidak pernah diproses Crash setelah diproses tetapi sebelum offset dicatat: pesan diproses lagi

Peran checkpoint. Checkpoint pada Structured Streaming menyimpan offset dan status kemajuan di lokasi yang tahan gagal. Sebelum sebuah micro-batch diproses, rentang offset-nya dicatat (write-ahead log), dan setelah batch selesai dicatat sebagai commit.

  • Mencegah kehilangan data. Bila sistem mati di tengah batch (offset tercatat tetapi belum di-commit), setelah restart Spark memproses ulang persis rentang offset yang sama, sehingga tidak ada data yang terlewat. Ini memerlukan sumber yang dapat dibaca ulang (file atau Kafka).
  • Mencegah pemrosesan berulang. Bila batch sudah di-commit, setelah restart Spark melanjutkan dari offset berikutnya dan tidak membaca ulang data lama. Hal ini terbukti pada C4: menjalankan ulang query dengan checkpoint yang sama menghasilkan nol batch baru dan jumlah baris Parquet tidak berubah.

Namun, pemrosesan ulang batch yang gagal berarti checkpoint saja setara at-least-once pada sisi keluaran. Untuk mendapatkan efek exactly-once (hasil akhir sama seperti diproses tepat satu kali), keluaran harus idempoten. Pada C4 hal ini dicapai dengan menulis setiap batch ke subfolder batch_id=<n> dengan mode overwrite, sehingga pemutaran ulang batch yang sama menimpa hasil lama dan tidak menggandakannya.

6 C1 – Kueri dan Agregasi NoSQL

6.1 Tujuan

Memanfaatkan fleksibilitas dokumen MongoDB untuk (a) menyaring merchant dengan beberapa syarat sekaligus, termasuk syarat pada field array (promo), (b) menghitung dokumen yang memiliki field opsional (jam_buka), dan (c) merangkum data per kategori dengan aggregation pipeline.

6.2 Implementasi

6.2.1 C1(a): Merchant Kuliner di Bandung dengan Promo Cashback

filter_c1a = {
    "kategori": "Kuliner",
    "kota": "Bandung",
    "promo": "cashback",            # pada field array, nilai tunggal berarti "array MENGANDUNG elemen ini"
    "rating": {"$gte": 4.0},        # $gte = greater than or equal
}
hasil_c1a = list(db["merchant"].find(filter_c1a, {"_id": 0, "nama": 1, "rating": 1, "promo": 1}).sort("nama", 1))
print(f"C1(a) Jumlah merchant yang memenuhi seluruh kondisi: {len(hasil_c1a)}")
## C1(a) Jumlah merchant yang memenuhi seluruh kondisi: 1
print(pd.DataFrame(hasil_c1a).to_string(index=False))
##        nama  rating      promo
## Merchant 75     4.1 [cashback]

6.2.2 C1(b): Merchant dengan Field jam_buka

jumlah_jam_buka = db["merchant"].count_documents({"jam_buka": {"$exists": True}})   # $exists menguji keberadaan field
print(f"\nC1(b) Merchant yang memiliki field jam_buka: {jumlah_jam_buka} dari {db['merchant'].count_documents({})}")
## 
## C1(b) Merchant yang memiliki field jam_buka: 61 dari 200

6.2.3 C1(c): Agregasi per Kategori

pipeline_c1c = [
    {"$group": {"_id": "$kategori",
                "jumlah_merchant": {"$sum": 1},
                "rata_rata_rating": {"$avg": "$rating"}}},
    {"$sort": {"rata_rata_rating": -1}},
]
c1c = pd.DataFrame(list(db["merchant"].aggregate(pipeline_c1c))).rename(columns={"_id": "kategori"})
c1c["rata_rata_rating"] = c1c["rata_rata_rating"].round(3)
c1c_d = c1c.to_dict("list")            # versi dict/list agar mudah dibaca R pada narasi interpretasi
print("\nC1(c) Agregasi per kategori:")
## 
## C1(c) Agregasi per kategori:
print(c1c.to_string(index=False))
##  jumlah_merchant  rata_rata_rating     kategori
##               28             4.146      Kuliner
##               42             4.117    Kesehatan
##               48             4.098 Transportasi
##               44             3.964      Hiburan
##               38             3.900       Retail

6.2.4 Penjelasan Teknis

  • find(filter, proyeksi) menyaring dokumen. Beberapa pasangan field–nilai dalam satu filter digabung dengan logika AND. Syarat "promo": "cashback" pada field array berarti “array promo mengandung elemen cashback”, tanpa perlu $in atau $elemMatch. Operator $gte dipakai untuk rating minimal 4,0. Proyeksi membatasi field yang dikirim ke klien.
  • count_documents({"jam_buka": {"$exists": True}}) menghitung dokumen yang memiliki field tersebut. Ini bekerja karena MongoDB berskema fleksibel: dokumen tanpa jam_buka tidak dianggap salah, hanya tidak memiliki field itu.
  • Pipeline $group lalu $sort memproses dokumen secara bertahap. $group mengelompokkan menurut $kategori lalu menghitung $sum: 1 (jumlah dokumen) dan $avg (rata-rata rating). $sort dengan nilai -1 mengurutkan menurun menurut rata-rata rating.

6.3 Hasil dan Interpretasi

C1(a). Hanya 1 dari 200 merchant yang memenuhi keempat syarat sekaligus: Merchant 75 (rating 4,1). Hasil yang sedikit ini wajar: 200 merchant tersebar pada 25 kombinasi kota–kategori (rata-rata 8 merchant per kombinasi), lalu masih disaring oleh syarat promo dan rating. Bagi BayarYuk, merchant ini adalah kandidat paling kuat untuk kampanye cashback kuliner di Bandung karena sudah berating baik dan sudah memiliki promo tersebut.

C1(b). Sebanyak 61 dari 200 merchant (30%) memiliki field jam_buka. Di basis data relasional, field yang hanya terisi sebagian akan menjadi kolom berisi NULL untuk 70% baris. Di MongoDB, dokumen yang tidak memiliki field itu cukup tidak menyimpannya. Konsekuensinya bagi aplikasi: jam operasional hanya diketahui untuk sebagian merchant, sehingga fitur seperti “sedang buka” harus menangani ketiadaan field tersebut.

C1(c). Rata-rata rating tertinggi dimiliki kategori Kuliner (4,146) dan terendah Retail (3,900), selisih hanya 0,25 poin. Dengan 28–48 merchant per kategori, urutan ini bersifat deskriptif dan belum cukup untuk menyimpulkan bahwa satu kategori lebih berkualitas daripada kategori lain.

7 C2 – Pemrosesan Log dengan Spark RDD

7.1 Tujuan

Mengolah 400.000 baris log teks secara terdistribusi: membaca, mem-parsing menjadi tuple bertipe, membuang baris rusak sambil menghitungnya dengan accumulator, lalu menghitung agregasi (jumlah per status dan tingkat kegagalan per merchant) dengan reduceByKey tanpa menarik data mentah ke driver.

7.2 Implementasi

7.2.1 Membaca dan Mem-parsing Log

from operator import add
from pyspark import StorageLevel

# accumulator untuk menghitung baris rusak
akum_rusak = sc.accumulator(0)

def parse_baris(baris):
    """Ubah satu baris log menjadi tuple
    (waktu, id_trx, id_user, id_merchant, nominal, status).

    Baris yang tidak sesuai format dibuang dan dicatat di accumulator.
    """
    f = baris.split("|")

    # jumlah kolom harus tepat 6
    if len(f) != 6:
        akum_rusak.add(1)
        return []

    try:
        waktu = datetime.strptime(f[0], "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)
        id_trx = f[1]
        id_user = f[2]
        id_merchant = f[3]
        nominal = int(f[4])
        status = f[5].strip().upper()

        # status di luar tiga kategori yang dipakai dianggap rusak
        if status not in ["SUKSES", "GAGAL", "PENDING"]:
            akum_rusak.add(1)
            return []

    except (ValueError, TypeError):
        akum_rusak.add(1)
        return []

    return [(waktu, id_trx, id_user, id_merchant, nominal, status)]

rdd_mentah = sc.textFile(jalur_log)
jumlah_mentah = rdd_mentah.count()

# hasil parsing disimpan agar dipakai ulang oleh aksi berikutnya
rdd_valid = rdd_mentah.flatMap(parse_baris).persist(StorageLevel.MEMORY_AND_DISK)

# aksi pertama menjalankan parsing; accumulator dibaca setelah aksi selesai
jumlah_valid = rdd_valid.count()
jumlah_rusak = akum_rusak.value
persen_rusak = jumlah_rusak / jumlah_mentah * 100 if jumlah_mentah > 0 else 0

print(f"Baris mentah (textFile)   : {jumlah_mentah:,}")
## Baris mentah (textFile)   : 400,000
print(f"Baris valid hasil parsing : {jumlah_valid:,}")
## Baris valid hasil parsing : 399,920
print(f"Baris rusak (accumulator) : {jumlah_rusak:,}")
## Baris rusak (accumulator) : 80
print(f"Persentase baris rusak    : {persen_rusak:.2f}%")
## Persentase baris rusak    : 0.02%
# satu contoh tuple diambil dari RDD yang sudah di-persist (tidak memicu parsing ulang)
contoh_tuple = rdd_valid.first()

print("\nContoh tuple hasil parsing:")
## 
## Contoh tuple hasil parsing:
print(contoh_tuple)
## (datetime.datetime(2026, 9, 20, 1, 13, 10, tzinfo=datetime.timezone.utc), 'T0000001', 'U42902', 'M172', 25000, 'SUKSES')

7.2.2 Jumlah Transaksi per Status

# pasangan (status, 1) dijumlahkan per status dengan reduceByKey
rdd_status = rdd_valid.map(lambda x: (x[5], 1))
hasil_status = rdd_status.reduceByKey(add).collect()    # hanya tiga baris yang dibawa ke driver
status_dict = dict(hasil_status)

print("Distribusi status transaksi:")
## Distribusi status transaksi:
for status in ["SUKSES", "GAGAL", "PENDING"]:
    jumlah = status_dict.get(status, 0)
    persentase = jumlah / jumlah_valid * 100 if jumlah_valid > 0 else 0
    print(f"{status:<10}: {jumlah:>10,} ({persentase:.2f}%)")
## SUKSES    :    371,870 (92.99%)
## GAGAL     :     20,166 (5.04%)
## PENDING   :      7,884 (1.97%)
jumlah_gagal = status_dict.get("GAGAL", 0)
gagal_global = jumlah_gagal / jumlah_valid * 100 if jumlah_valid > 0 else 0

print(f"\nTingkat kegagalan keseluruhan: {gagal_global:.2f}%")
## 
## Tingkat kegagalan keseluruhan: 5.04%

7.2.3 Tingkat Kegagalan per Merchant

# (id_merchant, (jumlah_transaksi, jumlah_gagal)), lalu dijumlahkan per merchant
rdd_merchant = rdd_valid.map(lambda x: (x[3], (1, 1 if x[5] == "GAGAL" else 0)))

hasil_merchant = rdd_merchant.reduceByKey(lambda a, b: (a[0] + b[0], a[1] + b[1]))

# hanya merchant dengan minimal 1.500 transaksi
merchant_aktif = hasil_merchant.filter(lambda x: x[1][0] >= 1500)

# tingkat kegagalan (%) tiap merchant
hasil_gagal_merchant = merchant_aktif.map(
    lambda x: (x[0], x[1][0], x[1][1], (x[1][1] / x[1][0]) * 100)
)

# lima merchant dengan tingkat kegagalan tertinggi
top5_merchant = hasil_gagal_merchant.takeOrdered(5, key=lambda x: -x[3])

n_merchant_total = hasil_merchant.count()
n_merchant_lolos = merchant_aktif.count()
print(f"Merchant pada log: {n_merchant_total} | memenuhi syarat minimal 1.500 transaksi: {n_merchant_lolos}")
## Merchant pada log: 200 | memenuhi syarat minimal 1.500 transaksi: 200
# nama merchant diambil dari simulasi MongoDB (mongomock), hanya untuk lima baris hasil
dok_merchant = db["merchant"].find({}, {"_id": 1, "nama": 1})
nama_merchant = {str(d["_id"]): d.get("nama", str(d["_id"])) for d in dok_merchant}

data_top5 = []
for id_merchant, total, gagal, tingkat in top5_merchant:
    data_top5.append({
        "id_merchant": id_merchant,
        "nama": nama_merchant.get(str(id_merchant), str(id_merchant)),
        "total_transaksi": total,
        "jumlah_gagal": gagal,
        "tingkat_gagal_%": tingkat,
    })

top5_df = pd.DataFrame(data_top5)

if not top5_df.empty:
    top5_df["tingkat_gagal_%"] = top5_df["tingkat_gagal_%"].round(2)

print("\nLima merchant dengan tingkat kegagalan tertinggi (minimal 1.500 transaksi):")
## 
## Lima merchant dengan tingkat kegagalan tertinggi (minimal 1.500 transaksi):
if not top5_df.empty:
    print(top5_df.to_string(index=False))
else:
    print("Tidak ditemukan merchant dengan minimal 1.500 transaksi.")
## id_merchant         nama  total_transaksi  jumlah_gagal  tingkat_gagal_%
##        M164 Merchant 164             2051           130             6.34
##        M048  Merchant 48             2025           125             6.17
##        M186 Merchant 186             1948           120             6.16
##        M147 Merchant 147             1959           120             6.13
##        M145 Merchant 145             1981           119             6.01
# versi dict/list untuk dibaca R pada bagian interpretasi
top5_d = top5_df.to_dict("list")

# kontribusi lima merchant tersebut terhadap seluruh transaksi dan transaksi gagal
if not top5_df.empty and jumlah_valid > 0 and jumlah_gagal > 0:
    total_trx_top5 = int(top5_df["total_transaksi"].sum())
    total_gagal_top5 = int(top5_df["jumlah_gagal"].sum())
    porsi_trx_top5 = total_trx_top5 / jumlah_valid * 100
    porsi_gagal_top5 = total_gagal_top5 / jumlah_gagal * 100
else:
    total_trx_top5 = 0
    total_gagal_top5 = 0
    porsi_trx_top5 = 0
    porsi_gagal_top5 = 0

print("\nKontribusi lima merchant teratas:")
## 
## Kontribusi lima merchant teratas:
print(f"Total transaksi             : {total_trx_top5:,}")
## Total transaksi             : 9,964
print(f"Total transaksi gagal       : {total_gagal_top5:,}")
## Total transaksi gagal       : 614
print(f"Porsi dari seluruh transaksi: {porsi_trx_top5:.2f}%")
## Porsi dari seluruh transaksi: 2.49%
print(f"Porsi dari transaksi gagal  : {porsi_gagal_top5:.2f}%")
## Porsi dari transaksi gagal  : 3.04%
# seberapa jauh merchant teratas menyimpang bila semua merchant sebenarnya memiliki tingkat kegagalan yang sama
rata_trx_merchant = float(jumlah_valid / n_merchant_total)
se_gagal = float(((gagal_global / 100) * (1 - gagal_global / 100) / rata_trx_merchant) ** 0.5 * 100)
z_top1 = float((top5_df["tingkat_gagal_%"].iloc[0] - gagal_global) / se_gagal) if not top5_df.empty else 0.0
print(f"\nGalat baku tingkat kegagalan per merchant bila semua merchant sama: {se_gagal:.2f} poin persentase")
## 
## Galat baku tingkat kegagalan per merchant bila semua merchant sama: 0.49 poin persentase
print(f"Merchant teratas berada {z_top1:.1f} galat baku di atas tingkat kegagalan keseluruhan")
## Merchant teratas berada 2.7 galat baku di atas tingkat kegagalan keseluruhan

7.2.4 Penjelasan Teknis

flatMap dipilih karena satu baris dapat menghasilkan nol (baris rusak) atau satu tuple, sedangkan map selalu menghasilkan tepat satu. Accumulator hanya dapat dipercaya jika dibaca di driver setelah action selesai. Pembaruan di dalam transformation dapat terhitung ganda bila tahap tersebut dihitung ulang, sehingga RDD disimpan dengan persist dan hanya satu action pertama (count) yang menjalankan parsing. Agregasi memakai reduceByKey karena penjumlahan dilakukan di tiap partisi sebelum shuffle (map-side combine), sedangkan groupByKey mengirim seluruh nilai lintas jaringan. Data yang dibawa ke driver hanya tiga baris status, lima baris top-5, dan beberapa hitungan.

7.3 Hasil dan Interpretasi

Kualitas parsing. Dari 400.000 baris mentah, 80 baris dibuang oleh accumulator (0,020%), sehingga 399.920 baris valid tersisa. Angka 80 sama persis dengan 400.000 ÷ 5.000, sesuai ketentuan bahwa satu baris rusak muncul setiap 5.000 baris (baris 1, 5.001, dan seterusnya), jadi tidak ada baris rusak yang lolos dan tidak ada baris sehat yang terbuang. Kehilangan data sekecil ini tidak mengubah hasil analisis secara berarti.

Status transaksi. SUKSES 371.870 (92,99%), GAGAL 20.166 (5,04%), dan PENDING 7.884 (1,97%). Satu dari kira-kira 20 transaksi berakhir gagal. Angka ini murni berdasarkan kolom status; hubungannya dengan nominal -1 (kesalahan sistem) dibahas pada C5.

Tingkat kegagalan per merchant. Dari 200 merchant, 200 memenuhi syarat minimal 1.500 transaksi, syarat yang mencegah tingkat kegagalan merchant bertransaksi sedikit menyesatkan. Lima merchant tertinggi memiliki tingkat kegagalan 6,01% sampai 6,34%, yaitu 1,2 sampai 1,3 kali tingkat keseluruhan, dan hanya menyumbang 3,0% dari seluruh transaksi gagal (2,5% dari seluruh transaksi). Selisih ini kecil: dengan rata-rata sekitar 2.000 transaksi per merchant, galat baku akibat kebetulan saja sekitar 0,49 poin, sehingga merchant teratas hanya 2,7 galat baku di atas rata-rata, wajar sebagai yang tertinggi di antara 200 merchant. Kelimanya (Merchant 164, Merchant 48, Merchant 186, Merchant 147, Merchant 145) dapat dipantau, tetapi data belum menunjukkan merchant yang benar-benar bermasalah.

8 C3 – Spark DataFrame dan SQL

8.1 Tujuan

Mengubah RDD hasil C2 menjadi DataFrame bertipe jelas, menggabungkannya dengan dimensi merchant dari simulasi MongoDB (mongomock), lalu menjawab pertanyaan bisnis: kategori mana yang memberi nominal terbesar di tiap kota dan seberapa dominan kategori tersebut.

8.2 Implementasi

8.2.1 DataFrame Transaksi dan Merchant

# Arrow dinonaktifkan agar toPandas() tidak membutuhkan PyArrow
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "false")

# C3(a): RDD hasil C2 diubah menjadi DataFrame dengan skema eksplisit
schema_trx = T.StructType([
    T.StructField("waktu", T.TimestampType(), True),
    T.StructField("id_trx", T.StringType(), True),
    T.StructField("id_user", T.StringType(), True),
    T.StructField("id_merchant", T.StringType(), True),
    T.StructField("nominal", T.LongType(), True),
    T.StructField("status", T.StringType(), True),
])

trx_df = spark.createDataFrame(rdd_valid, schema=schema_trx).cache()

print("Skema DataFrame transaksi:")
## Skema DataFrame transaksi:
trx_df.printSchema()
## root
##  |-- waktu: timestamp (nullable = true)
##  |-- id_trx: string (nullable = true)
##  |-- id_user: string (nullable = true)
##  |-- id_merchant: string (nullable = true)
##  |-- nominal: long (nullable = true)
##  |-- status: string (nullable = true)
jumlah_trx_awal = trx_df.count()
print(f"Jumlah baris trx_df: {jumlah_trx_awal:,}")
## Jumlah baris trx_df: 399,920
# C3(b): DataFrame merchant dari simulasi MongoDB (mongomock), hanya kolom _id, kategori, dan kota
merchant_docs = list(db["merchant"].find({}, {"_id": 1, "kategori": 1, "kota": 1}))

merchant_data = [
    (str(doc["_id"]), doc.get("kategori"), doc.get("kota"))
    for doc in merchant_docs
]

schema_merchant = T.StructType([
    T.StructField("_id", T.StringType(), False),
    T.StructField("kategori", T.StringType(), True),
    T.StructField("kota", T.StringType(), True),
])

merchant_df = spark.createDataFrame(merchant_data, schema=schema_merchant).cache()

print("\nSkema DataFrame merchant:")
## 
## Skema DataFrame merchant:
merchant_df.printSchema()
## root
##  |-- _id: string (nullable = false)
##  |-- kategori: string (nullable = true)
##  |-- kota: string (nullable = true)
jumlah_merchant = merchant_df.count()
print(f"Jumlah baris merchant_df: {jumlah_merchant:,}")
## Jumlah baris merchant_df: 200

8.2.2 Filter Transaksi dan Broadcast Join

# hanya transaksi SUKSES dengan 0 < nominal < Rp50.000.000
trx_filtered = trx_df.filter(
    (F.col("status") == "SUKSES") &
    (F.col("nominal") > 0) &
    (F.col("nominal") < 50000000)
)

# merchant (kecil) di-broadcast ke seluruh executor
trx_merchant = trx_filtered.join(
    broadcast(merchant_df),
    trx_filtered.id_merchant == merchant_df["_id"],
    "inner"
)

n_analisis = trx_merchant.count()
print(f"Jumlah transaksi setelah filter: {n_analisis:,}")
## Jumlah transaksi setelah filter: 371,658
# rencana eksekusi diperiksa, tetapi tidak dicetak seluruhnya
physical_plan = trx_merchant._jdf.queryExecution().executedPlan().toString()
broadcast_terpakai = "broadcasthashjoin" in physical_plan.lower()

print("\nOptimasi proses join:")
## 
## Optimasi proses join:
print("Metode join            :", "BroadcastHashJoin" if broadcast_terpakai else "metode join lainnya")
## Metode join            : BroadcastHashJoin
print("Kolom penghubung       : id_merchant = _id")
## Kolom penghubung       : id_merchant = _id
print("Jenis join             : inner join")
## Jenis join             : inner join
print("Data yang di-broadcast : data merchant")
## Data yang di-broadcast : data merchant
print("Status optimasi        :", "berhasil digunakan" if broadcast_terpakai else "BroadcastHashJoin tidak terdeteksi")
## Status optimasi        : berhasil digunakan

8.2.3 Agregasi dan Peringkat Kategori per Kota

# C3(c): total nominal per kota dan kategori
total_kk = (
    trx_merchant
    .groupBy("kota", "kategori")
    .agg(
        F.sum("nominal").alias("total_nominal"),
        F.count("*").alias("jumlah_trx"),
    )
    .orderBy("kota", F.desc("total_nominal"))
)

tabel_kk_df = total_kk.toPandas()          # hasil agregasi saja (kombinasi kota-kategori)
tabel_kk_df["rata_per_trx"] = tabel_kk_df["total_nominal"] / tabel_kk_df["jumlah_trx"]
tabel_kk_d = tabel_kk_df.to_dict("list")   # versi dict/list untuk dibaca R pada interpretasi

print(f"Jumlah kombinasi kota-kategori  : {len(tabel_kk_df):,}")
## Jumlah kombinasi kota-kategori  : 25
print(f"Jumlah transaksi yang dianalisis: {n_analisis:,}")
## Jumlah transaksi yang dianalisis: 371,658
print("Rata-rata nominal per transaksi antar kombinasi:")
## Rata-rata nominal per transaksi antar kombinasi:
print(f"Rp {tabel_kk_df['rata_per_trx'].min():,.0f} s.d. Rp {tabel_kk_df['rata_per_trx'].max():,.0f}")
## Rp 58,671 s.d. Rp 61,953
tabel_tampil = tabel_kk_df[["kota", "kategori", "total_nominal", "jumlah_trx"]].copy()
tabel_tampil["total_nominal"] = tabel_tampil["total_nominal"].map(lambda x: f"{x:,.0f}")
tabel_tampil["jumlah_trx"] = tabel_tampil["jumlah_trx"].map(lambda x: f"{x:,}")

print("\nTotal nominal per kota dan kategori (Rp):")
## 
## Total nominal per kota dan kategori (Rp):
print(tabel_tampil.to_string(index=False))
##     kota     kategori total_nominal jumlah_trx
##  Bandung Transportasi 1,455,560,000     24,136
##  Bandung      Hiburan 1,100,050,900     18,439
##  Bandung       Retail 1,016,883,400     16,801
##  Bandung      Kuliner   898,654,300     14,752
##  Bandung    Kesehatan   668,446,100     11,133
##  Jakarta Transportasi 1,138,838,700     18,616
##  Jakarta      Hiburan   888,514,200     15,017
##  Jakarta    Kesehatan   778,065,700     12,987
##  Jakarta       Retail   556,442,400      9,327
##  Jakarta      Kuliner   214,176,200      3,630
## Makassar       Retail 1,344,025,800     22,364
## Makassar    Kesehatan 1,236,338,300     20,434
## Makassar      Hiburan 1,003,119,800     16,847
## Makassar      Kuliner   580,311,300      9,367
## Makassar Transportasi   557,000,900      9,180
##    Medan Transportasi 1,439,104,900     23,921
##    Medan    Kesehatan 1,008,722,600     16,756
##    Medan      Hiburan 1,005,036,300     16,890
##    Medan      Kuliner   805,227,200     13,057
##    Medan       Retail   776,654,600     12,986
## Surabaya    Kesehatan 1,000,164,700     16,629
## Surabaya      Hiburan   868,802,300     14,808
## Surabaya Transportasi   800,425,300     13,153
## Surabaya      Kuliner   677,970,800     11,188
## Surabaya       Retail   564,119,200      9,240
# C3(d): kategori dengan total nominal tertinggi di tiap kota (window function)
window_kota = Window.partitionBy("kota")
window_rank = Window.partitionBy("kota").orderBy(F.desc("total_nominal"), F.asc("kategori"))

hasil_top = (
    total_kk
    .withColumn("total_kota", F.sum("total_nominal").over(window_kota))
    .withColumn("kontribusi_pct", F.round(F.col("total_nominal") / F.col("total_kota") * 100, 2))
    .withColumn("peringkat", F.row_number().over(window_rank))
    .filter(F.col("peringkat") == 1)
    .select("kota", "kategori", "total_nominal", "total_kota", "kontribusi_pct")
    .orderBy("kota")
)

hasil_top_df = hasil_top.toPandas()
hasil_top_d = hasil_top_df.to_dict("list")

hasil_top_tampil = hasil_top_df.copy()
for kolom in ["total_nominal", "total_kota"]:
    hasil_top_tampil[kolom] = hasil_top_tampil[kolom].map(lambda x: f"{x:,.0f}")

print("\nKategori dengan total nominal tertinggi per kota:")
## 
## Kategori dengan total nominal tertinggi per kota:
print(hasil_top_tampil.to_string(index=False))
##     kota     kategori total_nominal    total_kota  kontribusi_pct
##  Bandung Transportasi 1,455,560,000 5,139,594,700           28.32
##  Jakarta Transportasi 1,138,838,700 3,576,037,200           31.85
## Makassar       Retail 1,344,025,800 4,720,796,100           28.47
##    Medan Transportasi 1,439,104,900 5,034,745,600           28.58
## Surabaya    Kesehatan 1,000,164,700 3,911,482,300           25.57

8.2.4 Verifikasi dengan Spark SQL

trx_merchant.createOrReplaceTempView("transaksi_merchant")

hasil_top_sql = spark.sql("""
    WITH agregasi AS (
        SELECT kota, kategori,
               SUM(nominal) AS total_nominal,
               COUNT(*) AS jumlah_trx
        FROM transaksi_merchant
        GROUP BY kota, kategori
    ),
    peringkat AS (
        SELECT kota, kategori, total_nominal,
               SUM(total_nominal) OVER (PARTITION BY kota) AS total_kota,
               ROW_NUMBER() OVER (
                   PARTITION BY kota
                   ORDER BY total_nominal DESC, kategori ASC
               ) AS peringkat
        FROM agregasi
    )
    SELECT kota, kategori, total_nominal, total_kota,
           ROUND(total_nominal / total_kota * 100, 2) AS kontribusi_pct
    FROM peringkat
    WHERE peringkat = 1
    ORDER BY kota
""")

# hasil DataFrame dan Spark SQL dibandingkan dua arah
beda_dataframe_sql = hasil_top.exceptAll(hasil_top_sql).count()
beda_sql_dataframe = hasil_top_sql.exceptAll(hasil_top).count()
sql_sama = beda_dataframe_sql == 0 and beda_sql_dataframe == 0

print("Hasil Spark SQL identik dengan window function DataFrame:", sql_sama)
## Hasil Spark SQL identik dengan window function DataFrame: True
kontribusi_min = hasil_top_df["kontribusi_pct"].min()
kontribusi_max = hasil_top_df["kontribusi_pct"].max()

print("\nRingkasan C3:")
## 
## Ringkasan C3:
print(f"Jumlah transaksi awal          : {jumlah_trx_awal:,}")
## Jumlah transaksi awal          : 399,920
print(f"Transaksi setelah filter       : {n_analisis:,}")
## Transaksi setelah filter       : 371,658
print(f"Jumlah merchant                : {jumlah_merchant:,}")
## Jumlah merchant                : 200
print(f"Jumlah kombinasi kota-kategori : {len(tabel_kk_df):,}")
## Jumlah kombinasi kota-kategori : 25
print(f"Kontribusi kategori teratas    : {kontribusi_min:.2f}% - {kontribusi_max:.2f}%")
## Kontribusi kategori teratas    : 25.57% - 31.85%
print("BroadcastHashJoin terpakai     :", broadcast_terpakai)
## BroadcastHashJoin terpakai     : True
print("Hasil SQL dan DataFrame sama   :", sql_sama)
## Hasil SQL dan DataFrame sama   : True

8.2.5 Penjelasan Teknis

  • Skema eksplisit. Spark tidak perlu memindai data untuk menebak tipe, tipe yang salah gagal sejak awal (bukan diam-diam menjadi string), dan kontrak data terdokumentasi. waktu bertipe timestamp memungkinkan fungsi waktu seperti hour(), sedangkan nominal bertipe long mencegah pembulatan dan menjamin penjumlahan bilangan bulat yang tepat.
  • Broadcast join. merchant_df hanya 200 baris, sedangkan transaksi ratusan ribu baris. Dengan broadcast(), tabel kecil disalin ke setiap executor sehingga tabel besar tidak perlu di-shuffle berdasarkan id_merchant. Rencana eksekusi di atas memastikan Spark benar-benar memakai BroadcastHashJoin.
  • Window function / SQL. Kita membutuhkan peringkat kategori di dalam tiap kota sekaligus total kota pada baris yang sama. GROUP BY biasa menghilangkan baris detail, sedangkan SUM() OVER (PARTITION BY kota) dan ROW_NUMBER() mempertahankan baris kategori dan menambahkan konteks kota. Persentase kontribusi dihitung sekali jalan tanpa self-join. row_number dipilih (bukan rank) agar tepat satu kategori terpilih per kota meskipun terjadi nilai seri.

8.3 Hasil dan Interpretasi

Struktur dan join. trx_df memiliki enam kolom dengan tipe sesuai skema (waktu timestamp, nominal long), dan rencana eksekusi memastikan join memakai BroadcastHashJoin. Dari 399.920 transaksi valid, 371.658 (92,9%) lolos filter analitik. Sisanya dikeluarkan karena berstatus GAGAL atau PENDING, bernominal -1, atau bernominal Rp250 juta. Kombinasi kota–kategori terisi lengkap (25 dari 25 kelompok).

Kategori teratas per kota. Kategori pemimpin tiap kota: Transportasi di Bandung; Transportasi di Jakarta; Retail di Makassar; Transportasi di Medan; Kesehatan di Surabaya. Kontribusinya terhadap total kota berkisar 25,57% (Surabaya) sampai 31,85% (Jakarta), sehingga kategori teratas hanya menyumbang sekitar seperempat sampai sepertiga nominal kota dan tidak ada kota yang bergantung pada satu kategori saja. Kombinasi terkecil adalah Kuliner di Jakarta (Rp214.176.200), jauh di bawah kombinasi terbesar (Rp1.455.560.000).

Apa yang menggerakkan peringkat. Rata-rata nominal per transaksi hampir seragam antar kombinasi (Rp58.671 sampai Rp61.953), sehingga peringkat total nominal terutama mengikuti jumlah transaksi, bukan besarnya nilai per transaksi. Dengan kata lain, kategori dominan adalah kategori yang paling sering dipakai pengguna. Hasil Spark SQL identik dengan hasil window function, sehingga kedua pendekatan dapat dipakai bergantian sesuai preferensi tim. Seluruh temuan berlaku untuk periode 1–30 September 2026 pada data simulasi.

9 C4 – Structured Streaming

9.1 Tujuan

Membangun proses ingestion dari landing zone yang berjalan per berkas, memisahkan event tidak valid (nominal null), menyimpan event valid ke Parquet, mencatat jumlah valid/tidak valid per batch, dan membuktikan bahwa checkpoint mencegah pemrosesan ulang.

9.2 Implementasi

9.2.1 Definisi Stream dan Fungsi Batch

skema_event = T.StructType([                       # skema eksplisit: wajib untuk streaming file
    T.StructField("id_trx",      T.StringType(),    True),
    T.StructField("id_merchant", T.StringType(),    True),
    T.StructField("nominal",     T.LongType(),      True),
    T.StructField("waktu",       T.TimestampType(), True),
])

folder_invalid = (BASE_DIR / "output" / "parquet_tidak_valid").as_posix()    # karantina event tidak valid
catatan_batch = []                                  # catatan pemrosesan tiap batch (dikumpulkan di driver)

def proses_batch(batch_df, batch_id):
    """Dipanggil Spark untuk setiap micro-batch (di sini: satu file JSON per batch)."""
    batch_df = batch_df.cache()
    valid   = batch_df.filter(F.col("nominal").isNotNull())
    invalid = batch_df.filter(F.col("nominal").isNull())
    # satu pemindaian: count("*") menghitung semua event, count("nominal") hanya yang tidak null
    n_total, n_valid = batch_df.select(F.count("*"), F.count("nominal")).first()
    n_invalid = n_total - n_valid
    # Penulisan per batch ke subfolder batch_id=<n> dengan mode overwrite bersifat idempoten:
    # bila batch yang sama diputar ulang setelah kegagalan, hasilnya tidak terduplikasi.
    valid.write.mode("overwrite").parquet(f"{folder_parquet}/batch_id={batch_id}")
    invalid.write.mode("overwrite").parquet(f"{folder_invalid}/batch_id={batch_id}")
    catatan_batch.append({"batch_id": batch_id, "event_masuk": n_total,
                          "event_valid": n_valid, "event_tidak_valid": n_invalid})
    batch_df.unpersist()

def jalankan_stream():
    aliran = (spark.readStream
                   .format("json")
                   .schema(skema_event)
                   .option("maxFilesPerTrigger", 1)       # maksimal 1 file per micro-batch
                   .option("timestampFormat", "yyyy-MM-dd HH:mm:ss")
                   .load(folder_landing))
    q = (aliran.writeStream
               .foreachBatch(proses_batch)
               .option("checkpointLocation", folder_checkpoint)   # menyimpan file mana yang sudah diproses
               .trigger(availableNow=True)                        # proses semua data yang tersedia lalu berhenti
               .start())
    q.awaitTermination()

9.2.2 Menjalankan Stream

jalankan_stream()

print("Catatan pemrosesan setiap batch:")
## Catatan pemrosesan setiap batch:
catatan_df = pd.DataFrame(catatan_batch)
catatan_d = catatan_df.to_dict("list")                      # untuk narasi interpretasi
print(catatan_df.to_string(index=False))
##  batch_id  event_masuk  event_valid  event_tidak_valid
##         0          500          486                 14
##         1          500          479                 21
##         2          500          482                 18
print("\nTotal: masuk =", catatan_df["event_masuk"].sum(), "| valid =", catatan_df["event_valid"].sum(),
      "| tidak valid =", catatan_df["event_tidak_valid"].sum())
## 
## Total: masuk = 1500 | valid = 1447 | tidak valid = 53

9.2.3 Verifikasi Parquet dan Uji Checkpoint

# baca ulang Parquet dari disk (bukan dari catatan di memori)
parquet_df = spark.read.parquet(folder_parquet)
total_parquet = parquet_df.count()
print("\nJumlah baris tersimpan di Parquet (dibaca ulang dari disk):", total_parquet)
## 
## Jumlah baris tersimpan di Parquet (dibaca ulang dari disk): 1447
print("Baris per batch di Parquet:", {r["batch_id"]: r["n"] for r in
      parquet_df.groupBy("batch_id").agg(F.count("*").alias("n")).orderBy("batch_id").collect()})
## Baris per batch di Parquet: {0: 486, 1: 479, 2: 482}
print("Jumlah baris di karantina (tidak valid)                  :", spark.read.parquet(folder_invalid).count())
## Jumlah baris di karantina (tidak valid)                  : 53
# pembanding independen: JSON dibaca secara batch, lalu nominal tidak null dihitung
valid_batch_read = spark.read.schema(skema_event).option("timestampFormat", "yyyy-MM-dd HH:mm:ss").json(folder_landing).filter(F.col("nominal").isNotNull()).count()
print("Pembanding (baca batch langsung dari JSON)             :", valid_batch_read,
      "| konsisten dengan Parquet:", valid_batch_read == total_parquet)
## Pembanding (baca batch langsung dari JSON)             : 1447 | konsisten dengan Parquet: True
# Uji checkpoint: query yang sama dijalankan ulang. Tidak ada file baru, jadi tidak boleh ada batch baru.
jumlah_catatan_sebelum = len(catatan_batch)
jalankan_stream()
print("\nUji checkpoint: batch baru setelah menjalankan ulang =", len(catatan_batch) - jumlah_catatan_sebelum,
      "| total Parquet sekarang =", spark.read.parquet(folder_parquet).count())
## 
## Uji checkpoint: batch baru setelah menjalankan ulang = 0 | total Parquet sekarang = 1447

9.2.4 Penjelasan Teknis

  • format("json") dengan schema(skema_event) membaca file JSON dengan tipe yang pasti. Streaming file memang mewajibkan skema agar tidak perlu menebak tipe pada data yang belum datang.
  • maxFilesPerTrigger=1 membatasi satu file per micro-batch, sehingga tiga file diproses sebagai tiga batch terpisah.
  • trigger(availableNow=True) memproses seluruh data yang ada lalu menghentikan query. Cocok untuk ingestion terjadwal (misalnya tiap jam) yang berperilaku seperti streaming tetapi tidak berjalan terus-menerus.
  • checkpointLocation menyimpan offset dan daftar file yang sudah diproses. Setelah restart, Spark melanjutkan dari titik terakhir: data tidak hilang (karena lanjut dari offset terakhir) dan tidak diproses ulang.
  • foreachBatch memberi akses penuh pada API DataFrame untuk satu batch, sehingga pemisahan valid/tidak valid, penulisan Parquet, dan pencatatan dapat dilakukan dalam satu fungsi.

9.3 Hasil dan Interpretasi

Per batch. Spark memproses tiga file sebagai tiga micro-batch (batch_id 0, 1, 2) yang masing-masing berisi 500 event. Jumlah event valid per batch adalah 486, 479, 482, sedangkan event tidak valid (nominal null) adalah 14, 21, 18. Total tidak valid 53 dari 1.500 event (3,5%), sesuai karakteristik dataset. Jumlah tidak valid antar batch bervariasi (14 sampai 21), tetapi dengan hanya tiga titik pengamatan perbedaan ini tidak dapat ditafsirkan sebagai tren. Pemantauan lebih panjang diperlukan sebelum menyimpulkan bahwa kualitas sumber berubah.

Verifikasi. Parquet dibaca ulang dari disk dan berisi 1.447 baris, sama dengan penjumlahan catatan batch dan sama dengan hitungan pembanding yang membaca JSON langsung secara batch. Karena 1.447 valid + 53 di karantina = 1.500, tidak ada event yang hilang. Event tidak valid tidak dibuang diam-diam, melainkan disimpan terpisah sehingga dapat ditelusuri (misalnya apakah nominal null berasal dari satu merchant tertentu).

Peran checkpoint. Saat query yang sama dijalankan ulang dengan checkpoint yang sama, tidak ada batch baru yang diproses dan jumlah baris Parquet tetap 1.447. Checkpoint mengingat file yang sudah diproses, sehingga ingestion terjadwal tidak membaca ulang data lama. Perlu dicatat bahwa checkpoint sendiri menjamin Spark tidak mengulang batch yang sudah selesai, sedangkan keamanan terhadap kegagalan di tengah batch datang dari penulisan idempoten per batch_id pada fungsi foreachBatch (lihat jawaban B4).

10 C5 – EDA Skala Besar dan Insight

10.1 Tujuan

Memeriksa kualitas data transaksi, memahami sebaran nominal tanpa menarik data mentah ke driver, dan menemukan pola waktu transaksi sukses. Dari analisis ini ditarik dua insight yang dapat ditindaklanjuti.

10.2 Implementasi

10.2.1 C5(1): Laporan Kualitas Data (Satu Operasi select)

# C5(1): satu select dengan agregasi kondisional (satu kali pemindaian data)
laporan = trx_df.select(
    F.count("*").alias("total_baris_valid"),
    F.sum(F.when(F.col("nominal") < 0, 1).otherwise(0)).alias("nominal_negatif"),
    F.sum(F.when(F.col("nominal") >= 50_000_000, 1).otherwise(0)).alias("nominal_min_50jt"),
    F.sum(F.when(F.col("status") == "PENDING", 1).otherwise(0)).alias("status_pending"),
).first().asDict()
for k, v in laporan.items():
    print(f"{k:20s}: {v:>9,d}  ({v / laporan['total_baris_valid']:.3%})")
## total_baris_valid   :   399,920  (100.000%)
## nominal_negatif     :       150  (0.038%)
## nominal_min_50jt    :        80  (0.020%)
## status_pending      :     7,884  (1.971%)

10.2.2 C5(2): Distribusi Nominal dengan approxQuantile

# C5(2): persentil nominal dengan approxQuantile. Parameter relativeError membatasi galat pada PERINGKAT data
# (bukan pada nilainya): makin kecil makin akurat tetapi makin mahal; 0 berarti hasil eksak.
peluang = [0.05, 0.50, 0.95, 0.99]
nama_p = ["P5", "P50", "P95", "P99"]

# persentil dihitung dari transaksi dengan nominal valid (0 < nominal < Rp50 juta), sama dengan filter pada C3
nominal_valid = trx_df.filter((F.col("nominal") > 0) & (F.col("nominal") < 50_000_000))
persentil = dict(zip(nama_p, nominal_valid.approxQuantile("nominal", peluang, 0.0001)))
print("Persentil nominal transaksi valid (relativeError = 0,0001):")
## Persentil nominal transaksi valid (relativeError = 0,0001):
for k, v in persentil.items():
    print(f"{k:4s}: Rp {v:>14,.0f}")
## P5  : Rp          9,200
## P50 : Rp         40,100
## P95 : Rp        176,000
## P99 : Rp        326,300
# Pembanding: statistik berbasis rata-rata sangat sensitif terhadap nilai ekstrem, persentil tidak.
dampak = trx_df.agg(
    F.avg("nominal").alias("rata2_mentah"),
    F.avg(F.when((F.col("nominal") > 0) & (F.col("nominal") < 50_000_000), F.col("nominal"))).alias("rata2_bersih"),
    F.max("nominal").alias("nominal_maks"),
    F.sum(F.when(F.col("status") == "SUKSES", F.col("nominal"))).alias("total_sukses_mentah"),
    F.sum(F.when((F.col("status") == "SUKSES") & (F.col("nominal") > 0) & (F.col("nominal") < 50_000_000),
                 F.col("nominal"))).alias("total_sukses_bersih"),
    F.sum(F.when((F.col("nominal") >= 50_000_000) & (F.col("status") == "SUKSES"), 1).otherwise(0)).alias("ekstrem_sukses"),
    F.sum(F.when((F.col("nominal") < 0) & (F.col("status") == "GAGAL"), 1).otherwise(0)).alias("negatif_gagal"),
    F.sum(F.when((F.col("nominal") < 0) & (F.col("status") == "SUKSES"), 1).otherwise(0)).alias("negatif_sukses"),
).first().asDict()
print(f"\nRata-rata nominal mentah  : Rp {dampak['rata2_mentah']:>12,.0f}")
## 
## Rata-rata nominal mentah  : Rp      110,155
print(f"Rata-rata nominal bersih  : Rp {dampak['rata2_bersih']:>12,.0f}   (0 < nominal < Rp50 juta)")
## Rata-rata nominal bersih  : Rp       60,180   (0 < nominal < Rp50 juta)
print(f"Nominal maksimum          : Rp {dampak['nominal_maks']:>12,.0f}")
## Nominal maksimum          : Rp  250,000,000
print(f"Total nominal SUKSES mentah: Rp {dampak['total_sukses_mentah']:>16,.0f}")
## Total nominal SUKSES mentah: Rp   41,632,655,765
print(f"Total nominal SUKSES bersih: Rp {dampak['total_sukses_bersih']:>16,.0f}")
## Total nominal SUKSES bersih: Rp   22,382,655,900
print(f"Selisih akibat nilai ekstrem: Rp {dampak['total_sukses_mentah'] - dampak['total_sukses_bersih']:>14,.0f} "
      f"({(dampak['total_sukses_mentah'] - dampak['total_sukses_bersih']) / dampak['total_sukses_mentah']:.1%} dari total mentah)")
## Selisih akibat nilai ekstrem: Rp 19,249,999,865 (46.2% dari total mentah)
print(f"Transaksi SUKSES bernominal >= Rp50 juta: {dampak['ekstrem_sukses']} | nominal negatif berstatus GAGAL: {dampak['negatif_gagal']} | berstatus SUKSES: {dampak['negatif_sukses']}")
## Transaksi SUKSES bernominal >= Rp50 juta: 77 | nominal negatif berstatus GAGAL: 9 | berstatus SUKSES: 135
dampak_r = {k: float(v) for k, v in dampak.items()}   # float agar angka > 2^31 tidak meluap saat dibaca R (integer 32-bit)
rasio_ekstrem_p99 = dampak["nominal_maks"] / persentil["P99"]                    # untuk narasi interpretasi
print(f"Nominal maksimum = {rasio_ekstrem_p99:,.0f} kali P99")
## Nominal maksimum = 766 kali P99

10.2.3 C5(3): Transaksi SUKSES Berdasarkan Jam

# C5(3): agregasi di Spark (24 baris), baru kemudian dibawa ke driver untuk grafik
per_jam = (trx_df.filter(F.col("status") == "SUKSES")
                 .groupBy(F.hour("waktu").alias("jam"))
                 .agg(F.count("*").alias("jumlah_sukses"))
                 .orderBy("jam")
                 .toPandas()
                 .set_index("jam").reindex(range(24), fill_value=0))       # pastikan 24 jam tampil
total_sukses = int(per_jam["jumlah_sukses"].sum())
per_jam["persen"] = (100 * per_jam["jumlah_sukses"] / total_sukses).round(2)
per_jam_d = per_jam.reset_index().to_dict("list")            # untuk narasi interpretasi (urut jam 0-23)
print(per_jam.T.to_string())
## jam                  0         1         2        3         4         5         6         7         8         9         10        11        12        13        14        15        16        17        18        19        20        21        22        23
## jumlah_sukses  15591.00  15456.00  15583.00  15246.0  15394.00  15780.00  15691.00  15340.00  15451.00  15574.00  15701.00  15503.00  15381.00  15478.00  15468.00  15358.00  15494.00  15536.00  15450.00  15461.00  15507.00  15573.00  15543.00  15311.00
## persen             4.19      4.16      4.19      4.1      4.14      4.24      4.22      4.13      4.15      4.19      4.22      4.17      4.14      4.16      4.16      4.13      4.17      4.18      4.15      4.16      4.17      4.19      4.18      4.12
jam_puncak = int(per_jam["jumlah_sukses"].idxmax())
jam_sepi   = int(per_jam["jumlah_sukses"].idxmin())
tiga_teratas = per_jam["jumlah_sukses"].nlargest(3)
tiga_jam, tiga_total = [int(j) for j in tiga_teratas.index], int(tiga_teratas.sum())   # untuk narasi interpretasi
rasio_puncak = float(per_jam.loc[jam_puncak, "jumlah_sukses"] / per_jam.loc[jam_sepi, "jumlah_sukses"])
sd_jam  = float(per_jam["jumlah_sukses"].std())            # simpangan baku jumlah transaksi antar jam
sd_acak = float(per_jam["jumlah_sukses"].mean() ** 0.5)    # simpangan baku bila hanya ada fluktuasi acak (Poisson)
print(f"Simpangan baku antar jam: {sd_jam:,.0f} | bila hanya fluktuasi acak: {sd_acak:,.0f}")
## Simpangan baku antar jam: 126 | bila hanya fluktuasi acak: 124
print(f"\nJam puncak: {jam_puncak}:00 ({int(per_jam.loc[jam_puncak, 'jumlah_sukses']):,d} transaksi) | "
      f"jam tersepi: {jam_sepi}:00 ({int(per_jam.loc[jam_sepi, 'jumlah_sukses']):,d} transaksi)")
## 
## Jam puncak: 5:00 (15,780 transaksi) | jam tersepi: 3:00 (15,246 transaksi)
print(f"Rasio puncak/tersepi: {per_jam.loc[jam_puncak, 'jumlah_sukses'] / per_jam.loc[jam_sepi, 'jumlah_sukses']:.1f}x")
## Rasio puncak/tersepi: 1.0x
print("Tiga jam teratas:", ", ".join(f"{j}:00" for j in tiga_teratas.index),
      f"-> {100 * tiga_teratas.sum() / total_sukses:.1f}% dari seluruh transaksi sukses")
## Tiga jam teratas: 5:00, 10:00, 6:00 -> 12.7% dari seluruh transaksi sukses

Grafik batang berikut interaktif pada keluaran HTML: arahkan kursor ke batang untuk melihat jam, jumlah transaksi, dan persentasenya, seret area grafik untuk memperbesar (klik dua kali untuk mengembalikan), dan klik legenda untuk menyembunyikan atau menampilkan satu kelompok jam.

library(plotly)

# data hasil agregasi Spark (24 baris) dari chunk Python di atas
d_jam <- data.frame(
  jam           = unlist(py$per_jam_d$jam),
  jumlah_sukses = unlist(py$per_jam_d$jumlah_sukses),
  persen        = unlist(py$per_jam_d$persen)
)
d_jam$kelompok <- ifelse(d_jam$jam %in% unlist(py$tiga_jam), "Tiga jam tersibuk", "Jam lainnya")
d_jam$teks <- sprintf("Pukul %02d:00-%02d:59<br>Transaksi SUKSES: %s<br>Porsi: %s%%",
                      d_jam$jam, d_jam$jam, fmt(d_jam$jumlah_sukses), fmt(d_jam$persen, 2))

plot_ly(d_jam, x = ~jam, y = ~jumlah_sukses, color = ~kelompok,
        colors = c("Jam lainnya" = "#7f8c8d", "Tiga jam tersibuk" = "#c0392b"),
        type = "bar", text = ~teks, hoverinfo = "text") %>%
  layout(
    title   = list(text = "Transaksi SUKSES per jam (merah = tiga jam tersibuk)"),
    barmode = "overlay",
    xaxis   = list(title = "Jam (0-23)", dtick = 1),
    yaxis   = list(title = "Jumlah transaksi SUKSES", tickformat = ",d"),
    legend  = list(orientation = "h", y = -0.2)
  ) %>%
  config(displaylogo = FALSE)

10.3 Hasil dan Interpretasi

Laporan kualitas data. Dari 399.920 baris valid, terdapat 150 nominal negatif (0,038%), 80 nominal di atas atau sama dengan Rp50 juta (0,020%), dan 7.884 transaksi PENDING (1,97%). Masalah kualitas bernominal tergolong sangat sedikit secara persentase, tetapi seperti terlihat di Insight 2, dampaknya terhadap nilai agregat tidak kecil.

Distribusi nominal. Dari transaksi bernominal valid (0 < nominal < Rp50 juta), median sekitar Rp40.100, P95 sekitar Rp176.000, dan P99 sekitar Rp326.300. Selisih P50 terhadap P95 yang besar menunjukkan distribusi menjulur ke kanan: sebagian besar transaksi bernilai kecil dan sebagian kecil bernilai besar. approxQuantile mengorbankan sedikit ketelitian (dibatasi relativeError) demi komputasi yang jauh lebih murah daripada pengurutan penuh pada data miliaran baris. Rata-rata nominal mentah Rp110.155 sekitar 1,8 kali rata-rata bersih Rp60.180, sehingga ukuran transaksi tipikal lebih baik dilaporkan dengan median dan persentil.

Transaksi SUKSES per jam. Grafik hampir datar: jumlah transaksi sukses per jam berkisar 15.246 sampai 15.780, sehingga rasio jam tertinggi terhadap jam tersepi hanya 1,04 kali. Tabel di atas memuat jumlah dan persentase tiap jam, dan nilai yang sama muncul saat kursor diarahkan ke batang pada grafik interaktif. Warna merah menandai tiga jam tertinggi, tetapi selisihnya terhadap jam lain sangat kecil.

10.4 Insight dan rekomendasi

10.4.1 Insight 1 – Beban Transaksi Merata Sepanjang Hari

  • Temuan. Jam tertinggi adalah pukul 05:00 dengan 15.780 transaksi sukses, dan jam tersepi pukul 03:00 dengan 15.246 transaksi, selisih hanya 3,5%. Tiga jam teratas (5:00, 10:00, 6:00) menyumbang 12,7% dari seluruh transaksi sukses, nyaris sama dengan 12,5% (3 dari 24 jam) yang diharapkan bila beban benar-benar merata.
  • Bukti. Porsi tiap jam berkisar 4,10% sampai 4,24%, dibandingkan 4,17% bila merata sempurna. Simpangan baku jumlah antar jam (126) berada pada orde yang sama dengan simpangan baku yang diharapkan dari fluktuasi acak murni (124), sehingga perbedaan antar jam konsisten dengan kebetulan dan bukan pola jam sibuk.
  • Interpretasi. Berbeda dari aplikasi pembayaran sungguhan yang umumnya memiliki jam sibuk, data ini tidak memperlihatkan jam sibuk maupun jam sepi yang berarti. Hal ini sesuai dengan cara dataset dibangkitkan, yaitu waktu transaksi ditarik seragam sepanjang 30 hari, sehingga tidak ada alasan menyimpulkan bahwa jam tertentu lebih penting. Analisis ini juga belum memisahkan data menurut hari dalam seminggu atau kategori merchant.
  • Rekomendasi. (1) Atur kapasitas dengan patokan rata-rata beban ditambah cadangan, tidak perlu per jam. (2) Lakukan rilis dan perawatan secara bertahap (rolling), karena tidak ada jam yang jelas lebih sepi. (3) Ulangi analisis pada data nyata per hari dan kategori merchant sebelum menetapkan kebijakan.

10.4.2 Insight 2 – Nilai Ekstrem Menggelembungkan Hampir Separuh Nilai Transaksi Sukses

  • Temuan. Terdapat 80 transaksi bernominal Rp250 juta (0,020% dari data), dan 77 di antaranya berstatus SUKSES. Selain itu terdapat 150 transaksi bernominal -1, tetapi hanya 9 yang berstatus GAGAL, sedangkan 135 berstatus SUKSES dan sisanya PENDING.
  • Bukti. Total nominal SUKSES mentah Rp41.632.655.765 turun menjadi Rp22.382.655.900 setelah nilai ekstrem dikeluarkan, sehingga 46,2% dari total mentah berasal dari hanya 77 transaksi. Nominal maksimum sekitar 766 kali P99 (Rp326.300). Rata-rata nominal mentah Rp110.155 berbanding Rp60.180 setelah dibersihkan.
  • Interpretasi. Nilai Rp250 juta berada jauh di luar sebaran transaksi normal, sehingga lebih mungkin kesalahan input atau transaksi yang perlu diverifikasi daripada transaksi wajar. Bila tidak dibersihkan, laporan omzet dan rata-rata nilai transaksi akan menyesatkan manajemen. Nominal -1 adalah kode kesalahan sistem, bukan nilai uang. Karena sebagian besar justru berstatus SUKSES, kolom status saja tidak cukup untuk menyaring data bermasalah: 135 transaksi “sukses” tidak memiliki nilai uang yang sah. Sebaliknya, tingkat kegagalan pada C2 (5,04%) hampir tidak tercemar nominal -1, karena hanya 0,04% dari seluruh transaksi GAGAL yang bernominal -1.
  • Rekomendasi. (1) Validasi di titik masuk data: tolak atau tandai nominal ≤ 0 dan nominal di atas batas wajar (seperti karantina pada C4). (2) Tinjau manual 77 transaksi SUKSES bernominal Rp250 juta sebelum masuk laporan keuangan. (3) Pisahkan kode kesalahan sistem dari status bisnis, misalnya dengan kolom kode_error. (4) Laporkan nilai tipikal dengan median dan persentil, bukan rata-rata.

Catatan. Dataset pada dokumen ini adalah simulasi yang dibangkitkan dengan seed tetap mengikuti file acuan UTS. Anomali yang disuntikkan secara sengaja hanyalah nominal -1, nominal Rp250 juta, baris log rusak, dan nominal null pada data streaming. Jam transaksi dan merchant tidak diberi pola khusus, sehingga insight 1 justru menemukan beban yang merata. Yang divalidasi di sini adalah metode analisisnya, bukan perilaku pengguna BayarYuk yang sesungguhnya.

11 Kesimpulan

Dokumen ini membangun, menjalankan, dan menganalisis alur Big Data BayarYuk dari tiga sumber berbeda dengan teknologi yang sesuai bagi masing-masing sumber. Seluruh angka di bawah berasal dari eksekusi kode pada dokumen ini.

  1. NoSQL (C1). Simulasi MongoDB (mongomock) memudahkan penyaringan multi-syarat termasuk pada field array, menangani field opsional (jam_buka hanya pada 61 dari 200 merchant), dan merangkum data lewat aggregation pipeline. Hanya 1 merchant memenuhi seluruh syarat kampanye cashback kuliner di Bandung.
  2. Spark RDD (C2). Dari 400.000 baris log, 80 baris rusak terdeteksi oleh accumulator dan dibuang. Agregasi dikerjakan dengan reduceByKey tanpa menarik data mentah ke driver. Tingkat kegagalan keseluruhan 5,04% dan relatif seragam antar merchant: lima merchant tertinggi hanya menyumbang 3,0% dari seluruh transaksi gagal.
  3. DataFrame dan SQL (C3). Skema eksplisit dan broadcast join menghasilkan total nominal per kota dan kategori. Kategori teratas tiap kota menyumbang 25,6% sampai 31,9%, dan hasil window function sama dengan hasil Spark SQL.
  4. Structured Streaming (C4). Tiga file diproses sebagai tiga batch dengan checkpoint. Sebanyak 1.447 event valid tersimpan di Parquet dan 53 event tidak valid dikarantina, tanpa kehilangan data. Menjalankan ulang query tidak menambah batch baru.
  5. EDA skala besar (C5). Persentil, laporan kualitas, dan agregasi per jam dihitung di Spark, dan hanya hasil agregasi kecil yang dibawa ke driver. Dua insight utama: beban transaksi merata sepanjang hari tanpa jam sibuk yang jelas, dan 77 transaksi ekstrem menggelembungkan 46,2% dari nilai transaksi sukses.

Keterbatasan. Dataset adalah simulasi yang dibangkitkan dengan seed tetap, sehingga insight berlaku untuk data ini dan bukan untuk pengguna BayarYuk yang sebenarnya. Analisis juga baru mencakup satu bulan data (September 2026) dan belum memisahkan jam menurut hari atau kategori merchant. Perbedaan tingkat kegagalan antar merchant tidak cukup besar untuk menyimpulkan adanya merchant yang bermasalah.