Praktikum: Ujian Tengah Semester (UTS)
Big Data – Pertemuan 8: Evaluasi Materi Pertemuan 1–7
Nama: Nova Sitorus | NIM: 52240023
07 November 2026
Catatan render dan persiapan lingkungan: Rmd ini memakai chunk Python (
{python}) lewat paketreticulatedan menjalankan Apache Spark dalam mode lokal (local[*]) melalui PySpark, sehingga tidak membutuhkan klaster. Persiapan sekali saja: (1) di console R jalankaninstall.packages("reticulate"); (2) pasang Java JDK 17 atau 21 (mis. Eclipse Temurin darihttps://adoptium.net), lalu pastikan perintahjava -versiondi Terminal/Command Prompt menampilkan versinya – jika belum, atur variabel lingkunganJAVA_HOMEke folder instalasi JDK; (3) di Terminal jalankanpip install pyspark pandas numpy matplotlib pyarrow mongomock. Pengguna Windows yang mendapat peringatanwinutils.exedapat mengabaikannya selama chunk tetap berjalan, atau memakai WSL/Google Colab. Periksa Python yang dipakai R denganreticulate::py_config(). Semua dataset dibangkitkan di dalam kode, sehingga tidak memerlukan koneksi internet. Chunk bertandaeval=FALSEberisi perintah untuk klaster/server sungguhan (Hadoop, Kafka, dan sebagainya) dan tidak dijalankan saat knit.
Tautan video YouTube (Bagian D): (isi tautan Anda di sini)
1 Pendahuluan
Dokumen ini adalah laporan pengerjaan UTS Pengantar Big Data saya. Isinya jawaban untuk soal pilihan ganda dan esai, ditambah analisis studi kasus dompet digital BayarYuk yang saya kerjakan dengan MongoDB dan Apache Spark, mulai dari menanyai data merchant, mengolah log transaksi, sampai membaca data yang masuk secara streaming.
1.1 Latar Belakang
BayarYuk menyimpan datanya di tiga tempat yang berbeda. Data merchant (200 dokumen) ada di MongoDB, riwayat transaksi (400.000 baris) tersimpan sebagai berkas teks, dan transaksi baru datang lewat tiga berkas JSON di folder landing. Datanya juga belum bersih: ada baris log yang rusak, nominal bernilai -1, nominal Rp250 juta yang tidak wajar, dan sebagian event streaming tidak memiliki nominal. Karena itu, sebelum dianalisis data perlu dibersihkan dan divalidasi dulu, dan karena ukurannya besar, pengolahannya harus dilakukan di Spark, bukan ditarik utuh ke memori.
1.2 Tujuan
- Menjawab pertanyaan teori tentang konsep Big Data, NoSQL, HDFS, Spark, dan ingestion beserta alasannya.
- Mengambil dan merangkum data merchant dari MongoDB dengan query dan aggregation pipeline.
- Mengolah log transaksi dengan Spark RDD, lalu menganalisisnya dengan DataFrame dan Spark SQL.
- Membaca data streaming secara aman dengan checkpoint dan menyimpan hasilnya ke Parquet.
- Memeriksa kualitas data dan merumuskan insight beserta tindakan yang disarankan.
1.3 Alur Pengerjaan
| Bagian | Isi | Yang dikerjakan |
|---|---|---|
| A | Pilihan ganda | 10 soal beserta alasan singkat untuk tiap jawaban |
| B | Esai singkat | 4 esai tentang scale-out dan shuffle, pilihan NoSQL, lazy evaluation, dan jaminan pengiriman data |
| C | Studi kasus BayarYuk | C1 query MongoDB, C2 log dengan RDD, C3 DataFrame dan SQL, C4 streaming, C5 EDA dan insight |
| D | Video presentasi | Penjelasan pengerjaan Bagian C dan insight |
1.4 Alat yang Digunakan
Seluruh pengerjaan dilakukan di RStudio dengan R Markdown. Kode
ditulis dalam Python lewat paket reticulate dan dijalankan
memakai PySpark dalam mode lokal. Data merchant disimpan di simulator
MongoDB (mongomock), sedangkan pandas dan
matplotlib hanya dipakai untuk menampilkan hasil agregasi
yang kecil, seperti grafik transaksi per jam. Sepanjang pengerjaan, data
mentah tidak ditarik ke driver; yang dibawa hanya hasil
agregasi.
2 Bagian A: Pilihan Ganda
Pilih satu jawaban yang paling tepat.
A1. Data GPS 2 juta pengemudi ojek daring yang diperbarui setiap 3 detik paling menonjol pada karakteristik … (a) Variety (b) Velocity (c) Veracity (d) Value
Jawaban A1: (b) Velocity
Alasan: Datanya masuk terus-menerus, tiap 3 detik dari 2 juta pengemudi, jadi yang paling kelihatan itu kecepatan datangnya data, alias velocity.
A2. Basis data yang paling tepat untuk menyimpan sesi login dengan masa berlaku 30 menit adalah … (a) Neo4j (b) MongoDB (c) Redis (d) HDFS
Jawaban A2: (c) Redis
Alasan: Redis menyimpan data di memori sehingga cepat, dan bisa diberi TTL, jadi sesi login 30 menit akan terhapus sendiri begitu waktunya habis.
A3. Pada MongoDB, tahap aggregation
pipeline yang memecah elemen array menjadi dokumen terpisah adalah
… (a) $match (b) $group (c)
$unwind (d) $project
Jawaban A3: (c) $unwind
Alasan: $unwind itu yang “membongkar” array,
jadi tiap elemennya jadi dokumen sendiri-sendiri. Tiga opsi lainnya
fungsinya beda: $match buat menyaring, $group
buat mengelompokkan, $project buat memilih field.
A4. Berkas 700 MB diunggah ke HDFS dengan blok 128 MB dan replikasi 3. Jumlah blok dan ruang disk yang terpakai adalah … (a) 5 blok, 700 MB (b) 6 blok, 2.100 MB (c) 6 blok, 2.304 MB (d) 18 blok, 2.304 MB
Jawaban A4: (b) 6 blok, 2.100 MB
Alasan: 700 MB dibagi 128 MB hasilnya sekitar 5,5, jadi ada 5 blok penuh ditambah 1 blok sisa (60 MB), total 6 blok. Blok terakhir di HDFS tidak dipaksa jadi 128 MB, jadi ruang yang dipakai tinggal 700 MB x 3 replika = 2.100 MB.
A5. Komponen HDFS yang menyimpan metadata lokasi blok adalah … (a) DataNode (b) NameNode (c) Secondary NameNode (d) ResourceManager
Jawaban A5: (b) NameNode
Alasan: NameNode yang memegang “peta” atau metadata, yaitu blok mana ada di DataNode mana. DataNode sendiri cuma menyimpan isi blok datanya.
A6. Operasi Spark berikut yang merupakan
aksi adalah … (a) map (b)
filter (c) reduceByKey (d)
count
Jawaban A6: (d) count
Alasan: count memaksa Spark benar-benar
menjalankan perhitungan dan mengembalikan angkanya ke driver, jadi itu
aksi. Sisanya (map, filter,
reduceByKey) transformasi yang baru dijalankan nanti saat
ada aksi.
A7. Untuk menjumlahkan nilai per kunci pada pair RDD
besar, operasi yang paling efisien adalah … (a)
groupByKey().mapValues(sum) (b)
reduceByKey(lambda a, b: a + b) (c) collect()
lalu dijumlah di Python (d) sortByKey()
Jawaban A7: (b)
reduceByKey(lambda a, b: a + b)
Alasan: reduceByKey sudah menjumlahkan nilai di
tiap partisi dulu sebelum data dikirim antar-node, jadi yang di-shuffle
jauh lebih sedikit. groupByKey mengirim semua nilai mentah,
dan collect() bisa bikin memori driver penuh.
A8. Mengapa skema eksplisit lebih disarankan
daripada inferSchema=True pada pipeline produksi?
(a) Skema eksplisit membuat data terkompresi (b) Lebih cepat, tidak
salah tebak tipe, dan terdokumentasi (c) inferSchema tidak
dapat membaca CSV (d) Skema eksplisit wajib untuk Parquet
Jawaban A8: (b) Lebih cepat, tidak salah tebak tipe, dan terdokumentasi
Alasan: Dengan inferSchema, Spark harus membaca
data dulu untuk menebak tipe, dan tebakannya bisa meleset. Skema
eksplisit lebih cepat, tipenya pasti, dan sekaligus jadi dokumentasi
struktur datanya.
A9. Topic Kafka memiliki 4 partisi dan consumer group memiliki 6 konsumen. Berapa konsumen yang aktif membaca? (a) 2 (b) 4 (c) 6 (d) 24
Jawaban A9: (b) 4
Alasan: Dalam satu consumer group, satu partisi cuma boleh dibaca satu konsumen. Karena partisinya 4, hanya 4 konsumen yang aktif dan 2 sisanya nganggur.
A10. Cara yang benar membuat histogram kolom
jarak_km dari 2 miliar baris adalah … (a)
df.toPandas()["jarak_km"].hist() (b)
df.collect() lalu plt.hist (c) Menghitung
jumlah per bin dengan groupBy di Spark, lalu
menggambar hasilnya (d) Mengambil 10 baris pertama
Jawaban A10: (c) Menghitung jumlah per bin dengan
groupBy di Spark, lalu menggambar hasilnya
Alasan: Untuk 2 miliar baris, hitung dulu jumlah per bin di
Spark, lalu gambar hasilnya yang cuma beberapa baris.
toPandas() atau collect() bakal membawa
seluruh data ke driver dan memorinya tidak akan cukup.
3 Bagian B: Esai Singkat (20%)
B1. Jelaskan perbedaan scale-up dan scale-out, dan jelaskan mengapa fase shuffle pada MapReduce/Spark merupakan bagian yang paling mahal. (5 poin)
Jawaban B1. Scale-up artinya kita memperkuat satu mesin, misalnya menambah RAM, CPU, atau disk. Caranya gampang, tapi ada batasnya, makin mahal, dan kalau mesin itu mati semuanya ikut berhenti. Scale-out artinya kita menambah jumlah mesin (yang harganya biasa saja) lalu membagi data dan pekerjaan ke semuanya. Kapasitasnya bisa terus dikembangkan dan kalau satu node rusak, yang lain tetap jalan, walaupun jadi butuh koordinasi antar-node.
Shuffle adalah tahap ketika hasil dari fase map
dikelompokkan berdasarkan kunci lalu dikirim ke node yang akan
memprosesnya di tahap berikutnya. Tahap ini paling mahal karena datanya
harus ditulis ke disk, diserialisasi, dan dikirim lewat jaringan,
padahal ini jauh lebih lambat daripada menghitung di memori. Selain itu
tahap berikutnya harus menunggu semua tugas sebelumnya selesai, dan
kalau sebaran kuncinya tidak merata, ada node yang kerjanya jauh lebih
berat. Makanya kita sebisa mungkin mengurangi shuffle, misalnya dengan
reduceByKey atau broadcast join.
B2. Sebuah marketplace membutuhkan: (i) katalog produk dengan atribut berbeda per kategori, (ii) keranjang belanja sementara, (iii) rekomendasi “teman Anda juga membeli”. Pilih basis data NoSQL untuk masing-masing dan jelaskan alasannya. (5 poin)
Jawaban B2.
Katalog produk yang atributnya beda-beda per kategori cocok pakai MongoDB (document store). Tiap produk disimpan sebagai dokumen JSON yang strukturnya bebas, jadi baju bisa punya
ukurandanwarna, sedangkan laptop punyaRAMdanprosesor, tanpa harus mengubah skema atau menyisakan banyak kolom kosong.Keranjang belanja sementara cocok pakai Redis (key-value di memori). Aksesnya sangat cepat, kuncinya bisa berupa ID pengguna dengan isi keranjang sebagai nilainya, dan lewat TTL keranjang yang ditinggalkan akan hilang sendiri. Ini pas karena datanya memang hanya sementara.
Rekomendasi “teman Anda juga membeli” cocok pakai Neo4j (graph database). Intinya ada pada hubungan: siapa berteman dengan siapa dan siapa membeli apa. Menelusuri hubungan berlapis seperti itu jauh lebih ringan di graf daripada di basis data lain yang harus melakukan banyak join.
B3. Jelaskan konsep lazy evaluation dan lineage pada Spark, serta manfaat masing-masing. (5 poin)
Jawaban B3. Lazy evaluation berarti Spark
tidak langsung mengerjakan transformasi seperti map,
filter, atau join saat kita menuliskannya.
Spark hanya mencatat rencananya, dan baru benar-benar menghitung ketika
ada aksi seperti count, show, atau
write. Manfaatnya, Spark bisa melihat seluruh rencana dulu
lalu mengoptimalkannya, misalnya menyaring data sedini mungkin,
menggabungkan beberapa langkah jadi satu stage, dan hanya
membaca kolom yang diperlukan, jadi tidak ada kerja sia-sia.
Lineage adalah catatan riwayat bagaimana sebuah RDD atau DataFrame dibentuk, mulai dari data sumber lalu transformasi apa saja yang sudah dilalui. Manfaat utamanya untuk ketahanan terhadap kegagalan: kalau ada partisi yang hilang karena node bermasalah, Spark tinggal menghitung ulang partisi itu dari sumbernya mengikuti catatan tadi, tanpa perlu menyimpan salinan seluruh data.
B4. Jelaskan perbedaan jaminan pengiriman at-most-once dan at-least-once, dan bagaimana checkpoint pada Spark Structured Streaming membantu mencegah data hilang atau terproses ganda. (5 poin)
Jawaban B4. Pada at-most-once, setiap pesan dikirim paling banyak satu kali. Tidak akan ada duplikat, tetapi kalau terjadi gangguan, pesan bisa hilang karena tidak dikirim ulang. Pada at-least-once, pesan dikirim ulang sampai ada konfirmasi sehingga tidak ada yang hilang, tetapi pesan yang sama bisa terproses lebih dari sekali.
Checkpoint di Spark Structured Streaming bekerja seperti penanda progres: Spark mencatat sampai offset mana data sudah diproses dan batch mana yang sudah selesai (commit). Kalau kueri berhenti atau gagal lalu dijalankan lagi, Spark melanjutkan dari catatan itu. Data yang belum selesai diproses dibaca lagi sehingga tidak hilang, sedangkan batch yang sudah selesai tidak diulang sehingga tidak terproses ganda. Dengan sumber yang bisa dibaca ulang (seperti Kafka atau berkas) dan penulisan hasil yang aman diulang, hasilnya praktis setara exactly-once.
4 Bagian C: Praktik Coding – Studi Kasus Dompet Digital “BayarYuk” (60%)
BayarYuk adalah aplikasi dompet digital. Tim data diminta menganalisis data merchant (tersimpan di MongoDB), log transaksi (berkas teks), dan aliran transaksi baru (streaming). Jalankan chunk berikut untuk menyiapkan Spark dan membangkitkan dataset ujian.
# 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\\hp\\AppData\\Local\\R\\cache\\R\\RETICU~1\\uv\\cache\\ARCHIV~1\\2WB--L~1\\Scripts\\python.exe'os.environ.setdefault("PYSPARK_DRIVER_PYTHON", sys.executable)## 'C:\\Users\\hp\\AppData\\Local\\R\\cache\\R\\RETICU~1\\uv\\cache\\ARCHIV~1\\2WB--L~1\\Scripts\\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 4.3.0 | master: local[*] | jumlah core paralel: 4# Mengimpor pustaka yang dipakai selama ujian.
import numpy as np
import pandas as pd
import os, json, 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 (dokumen 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 MongoDB (simulator).
db = mongomock.MongoClient()["bayaryuk"]
db["merchant"].insert_many(merchant)## <mongomock.results.InsertManyResult object at 0x0000014B7E6F23D0># 2) Log transaksi (berkas teks, 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 baris log rusak.
baris[::5000] = ["ERROR koneksi timeout"] * len(baris[::5000])
jalur_log = os.path.join(folder, "transaksi.log")
with open(jalur_log, "w") as f:
f.write("\n".join(baris))## 21622886# 3) Landing zone untuk streaming (3 berkas JSON, 3% nominal null)
landing = os.path.join(folder, "landing"); os.makedirs(landing)
for b in range(3):
with open(os.path.join(landing, f"batch_{b}.json"), "w") 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))}
f.write(json.dumps(e) + "\n")## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 95
## 94
## 94
## 93
## 93
## 94
## 94
## 95
## 94
## 95
## 94
## 93
## 94
## 94
## 95
## 94
## 95
## 93
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 93
## 93
## 94
## 94
## 94
## 95
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 93
## 95
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 95
## 94
## 94
## 93
## 95
## 95
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 95
## 95
## 94
## 94
## 95
## 94
## 94
## 93
## 94
## 93
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 95
## 94
## 95
## 94
## 94
## 95
## 93
## 94
## 94
## 95
## 94
## 95
## 93
## 94
## 95
## 94
## 94
## 95
## 95
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 95
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 95
## 94
## 95
## 95
## 95
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 95
## 94
## 94
## 93
## 94
## 94
## 95
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 93
## 94
## 94
## 95
## 94
## 95
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 95
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 93
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 95
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 93
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 95
## 94
## 95
## 94
## 95
## 93
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 95
## 94
## 93
## 94
## 94
## 94
## 95
## 94
## 94
## 95
## 94
## 93
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 93
## 95
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 95
## 95
## 94
## 93
## 93
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 95
## 95
## 95
## 95
## 94
## 94
## 94
## 94
## 96
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 95
## 94
## 94
## 94
## 93
## 95
## 94
## 94
## 94
## 94
## 93
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 95
## 94
## 95
## 95
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 95
## 93
## 94
## 93
## 95
## 94
## 94
## 95
## 94
## 94
## 94
## 93
## 94
## 94
## 95
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 93
## 94
## 94
## 95
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 95
## 94
## 94
## 93
## 94
## 93
## 95
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 95
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 93
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 95
## 93
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 95
## 94
## 94
## 94
## 93
## 94
## 95
## 94
## 94
## 94
## 94
## 93
## 95
## 95
## 94
## 94
## 95
## 94
## 95
## 94
## 94
## 95
## 94
## 94
## 94
## 93
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 93
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 95
## 94
## 94
## 94
## 94
## 94
## 95
## 95
## 93
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 95
## 93
## 94
## 93
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 95
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 95
## 94
## 93
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 95
## 95
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 95
## 93
## 94
## 94
## 94
## 93
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 95
## 94
## 94
## 95
## 94
## 95
## 94
## 95
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 95
## 93
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 93
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 95
## 93
## 95
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 95
## 93
## 95
## 95
## 94
## 95
## 94
## 95
## 93
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 95
## 93
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 95
## 94
## 95
## 94
## 93
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 95
## 94
## 94
## 93
## 95
## 93
## 94
## 95
## 94
## 95
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 93
## 94
## 95
## 95
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 95
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 93
## 94
## 94
## 94
## 94
## 95
## 94
## 93
## 94
## 94
## 95
## 94
## 93
## 95
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 93
## 94
## 94
## 95
## 94
## 95
## 95
## 94
## 94
## 94
## 94
## 94
## 95
## 95
## 94
## 94
## 95
## 96
## 94
## 93
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 93
## 95
## 95
## 94
## 94
## 93
## 95
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 93
## 93
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 93
## 94
## 93
## 94
## 95
## 95
## 94
## 93
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 93
## 94
## 94
## 94
## 93
## 93
## 95
## 94
## 94
## 95
## 94
## 94
## 93
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 95
## 95
## 93
## 93
## 94
## 94
## 94
## 95
## 95
## 94
## 95
## 94
## 94
## 94
## 95
## 94
## 95
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 93
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 95
## 93
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 95
## 95
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 93
## 95
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 95
## 95
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 95
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 95
## 93
## 94
## 94
## 94
## 95
## 94
## 95
## 94
## 94
## 95
## 94
## 94
## 94
## 95
## 94
## 95
## 94
## 94
## 93
## 94
## 95
## 94
## 94
## 94
## 94
## 95
## 94
## 95
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 93
## 94
## 95
## 94
## 95
## 93
## 93
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 94
## 94
## 93
## 94
## 93
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 95
## 94
## 94
## 94
## 93
## 95
## 94
## 94
## 95
## 94
## 94
## 94
## 93# 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: 34.1 Tugas C1: Kueri dan Agregasi NoSQL (10 poin)
- Dengan MongoDB (
db["merchant"]), tampilkan nama merchant Kuliner di Bandung yang memiliki promo"cashback"dan rating \(\ge\) 4,0. (b) Hitung jumlah merchant yang memiliki fieldjam_buka. (c) Dengan aggregation pipeline, hitung jumlah merchant dan rata-rata rating per kategori, urutkan dari rata-rata rating tertinggi.
# (a) Merchant Kuliner di Bandung dengan promo "cashback" dan rating >= 4,0.
# Kondisi pada field array "promo" otomatis berarti "array mengandung elemen".
filter_a = {"kategori": "Kuliner", "kota": "Bandung",
"promo": "cashback", "rating": {"$gte": 4.0}}
print("C1(a) Merchant Kuliner di Bandung, promo cashback, rating >= 4,0:")## C1(a) Merchant Kuliner di Bandung, promo cashback, rating >= 4,0:for d in db["merchant"].find(filter_a, {"_id": 0, "nama": 1, "rating": 1}):
print(" ", d["nama"], "| rating:", d["rating"])## Merchant 75 | rating: 4.1# (b) Jumlah merchant yang memiliki field jam_buka (skema fleksibel) dengan $exists.
jml_jam_buka = db["merchant"].count_documents({"jam_buka": {"$exists": True}})
print("\nC1(b) Merchant dengan field jam_buka:", jml_jam_buka, "dari 200")##
## C1(b) Merchant dengan field jam_buka: 61 dari 200# (c) Aggregation pipeline: $group per kategori, lalu $sort rating menurun.
pipeline = [
{"$group": {"_id": "$kategori",
"jumlah_merchant": {"$sum": 1},
"rata_rating": {"$avg": "$rating"}}},
{"$sort": {"rata_rating": -1}},
]
print("\nC1(c) Jumlah merchant dan rata-rata rating per kategori:")##
## C1(c) Jumlah merchant dan rata-rata rating per kategori:for d in db["merchant"].aggregate(pipeline):
print(" %-13s jumlah = %2d | rata-rata rating = %.3f"
% (d["_id"], d["jumlah_merchant"], d["rata_rating"]))## Kuliner jumlah = 28 | rata-rata rating = 4.146
## Kesehatan jumlah = 42 | rata-rata rating = 4.117
## Transportasi jumlah = 48 | rata-rata rating = 4.098
## Hiburan jumlah = 44 | rata-rata rating = 3.964
## Retail jumlah = 38 | rata-rata rating = 3.900Tafsiran C1. Merchant Kuliner di Bandung dengan
promo cashback dan rating minimal 4,0 ternyata hanya satu, yaitu
Merchant 75 (rating 4,1). Ada 61 dari 200 merchant (sekitar 30%) yang
punya field jam_buka, jadi di MongoDB sebuah field memang
boleh hanya ada di sebagian dokumen. Dari sisi rating, Kuliner paling
tinggi (4,146), lalu Kesehatan (4,117), dan Retail paling rendah
(3,900).
4.2 Tugas C2: Pemrosesan Log dengan Spark RDD (12 poin)
- Baca
jalur_logsebagai RDD, urai setiap baris menjadi tuple(waktu, id_trx, id_user, id_merchant, nominal, status), buang baris rusak, dan hitung jumlah baris rusak dengan accumulator. (b) Hitung jumlah transaksi perstatusdenganreduceByKey. (c) Hitung tingkat kegagalan (persentase statusGAGAL) perid_merchant, lalu tampilkan 5 merchant dengan tingkat kegagalan tertinggi (minimal 1.500 transaksi).
from operator import add
# (a) Accumulator dipakai agar jumlah baris rusak terhitung di seluruh partisi
# tanpa menarik data ke driver.
rusak = sc.accumulator(0)
def urai(baris):
"""Mengurai satu baris log; baris rusak dibuang dan dihitung."""
p = baris.split("|")
if len(p) != 6: # format salah, mis. "ERROR koneksi timeout"
rusak.add(1)
return []
try:
return [(p[0], p[1], p[2], p[3], int(p[4]), p[5])]
except ValueError: # nominal bukan angka
rusak.add(1)
return []
# flatMap: baris rusak menghasilkan list kosong sehingga otomatis terbuang.
# cache() agar log tidak dibaca ulang dan accumulator tidak terhitung dua kali.
rdd = sc.textFile(jalur_log).flatMap(urai).cache()
# Accumulator baru berisi nilai setelah ada aksi; count() menjalankan eksekusi.
n_valid = rdd.count()
print("C2(a) Baris valid:", n_valid, "| baris rusak (accumulator):", rusak.value)## C2(a) Baris valid: 399920 | baris rusak (accumulator): 80# (b) reduceByKey menjumlahkan per status dengan combine lokal sebelum shuffle.
per_status = rdd.map(lambda t: (t[5], 1)).reduceByKey(add).collect() # hanya 3 baris
print("\nC2(b) Jumlah transaksi per status:")##
## C2(b) Jumlah transaksi per status:for st, jml in sorted(per_status):
print(" %-8s %7d" % (st, jml))## GAGAL 20166
## PENDING 7884
## SUKSES 371870# (c) Tingkat kegagalan per merchant: nilai (jumlah_gagal, jumlah_total) dijumlahkan.
per_merchant = (rdd.map(lambda t: (t[3], (1 if t[5] == "GAGAL" else 0, 1)))
.reduceByKey(lambda a, b: (a[0] + b[0], a[1] + b[1])))
top5 = (per_merchant.filter(lambda kv: kv[1][1] >= 1500) # minimal 1.500 transaksi
.map(lambda kv: (kv[0], kv[1][1], kv[1][0], round(100 * kv[1][0] / kv[1][1], 2)))
.takeOrdered(5, key=lambda x: -x[3])) # 5 tertinggi
print("\nC2(c) 5 merchant dengan tingkat kegagalan tertinggi (min. 1.500 transaksi):")##
## C2(c) 5 merchant dengan tingkat kegagalan tertinggi (min. 1.500 transaksi):print(" %-6s %10s %8s %10s" % ("merchant", "transaksi", "gagal", "gagal (%)"))## merchant transaksi gagal gagal (%)for m, tot, g, pct in top5:
print(" %-8s %9d %8d %10.2f" % (m, tot, g, pct))## M164 2051 130 6.34
## M048 2025 125 6.17
## M186 1948 120 6.16
## M147 1959 120 6.13
## M145 1981 119 6.01Tafsiran C2. Accumulator mencatat 80 baris rusak
(setiap baris ke-5.000), jadi tersisa 399.920 baris yang bisa dipakai.
Rinciannya SUKSES 371.870, GAGAL 20.166, dan PENDING 7.884. Saya pakai
reduceByKey karena nilai dijumlahkan dulu di tiap partisi
sebelum di-shuffle, sedangkan groupByKey mengirim semua
nilainya. Yang dibawa ke driver hanya hasil agregasi (3 status dan 5
merchant), bukan data mentah, jadi tidak perlu collect()
pada data besar. Lima merchant dengan kegagalan tertinggi ada di kisaran
6,0-6,3%, dengan M164 paling atas.
4.3 Tugas C3: Analisis dengan Spark DataFrame dan SQL (14 poin)
- Ubah RDD hasil C2 menjadi DataFrame dengan skema yang benar
(
waktubertipe timestamp,nominalbertipe long). (b) Buat DataFrame merchant dari MongoDB (hanya_id,kategori,kota) dan gabungkan dengan transaksi SUKSES yang nominalnya valid (0 < nominal < 50 juta) menggunakan broadcast join. (c) Hitung total nominal per kota per kategori. (d) Dengan window function atau Spark SQL, tampilkan kategori teratas di setiap kota beserta porsinya (%) terhadap total kota tersebut.
from datetime import datetime
from pyspark.sql.types import (StructType, StructField, StringType,
TimestampType, LongType)
# (a) Skema eksplisit: waktu bertipe timestamp, nominal bertipe long.
skema_trx = StructType([
StructField("waktu", TimestampType(), True),
StructField("id_trx", StringType(), True),
StructField("id_user", StringType(), True),
StructField("id_merchant", StringType(), True),
StructField("nominal", LongType(), True),
StructField("status", StringType(), True),
])
# String waktu diubah menjadi objek datetime agar cocok dengan TimestampType.
rdd_typed = rdd.map(lambda t: (datetime.strptime(t[0], "%Y-%m-%d %H:%M:%S"),
t[1], t[2], t[3], int(t[4]), t[5]))
df_trx = spark.createDataFrame(rdd_typed, skema_trx).cache() # dipakai ulang di C5
print("C3(a) Skema df_trx:")## C3(a) Skema df_trx:df_trx.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)print("Jumlah baris df_trx:", df_trx.count())## Jumlah baris df_trx: 399920# (b) DataFrame merchant (hanya _id, kategori, kota); data kecil (200 baris).
skema_m = StructType([StructField("id_merchant", StringType()),
StructField("kategori", StringType()),
StructField("kota", StringType())])
df_m = spark.createDataFrame(
[(d["_id"], d["kategori"], d["kota"])
for d in db["merchant"].find({}, {"_id": 1, "kategori": 1, "kota": 1})],
skema_m)
# Transaksi SUKSES dengan nominal valid (0 < nominal < 50 juta).
trx_valid = df_trx.filter((F.col("status") == "SUKSES") &
(F.col("nominal") > 0) &
(F.col("nominal") < 50_000_000))
# broadcast(): tabel merchant kecil disalin ke semua executor sehingga
# tabel transaksi besar tidak perlu di-shuffle.
joined = trx_valid.join(F.broadcast(df_m), on="id_merchant", how="inner")
print("\nC3(b) Rencana fisik (cari BroadcastHashJoin):")##
## C3(b) Rencana fisik (cari BroadcastHashJoin):joined.explain()## == Physical Plan ==
## AdaptiveSparkPlan isFinalPlan=false
## +- Project [id_merchant#3, waktu#0, id_trx#1, id_user#2, nominal#4L, status#5, kategori#197, kota#198]
## +- BroadcastHashJoin [id_merchant#3], [id_merchant#196], Inner, BuildRight, false, false
## :- Filter (((((isnotnull(status#5) AND isnotnull(nominal#4L)) AND (status#5 = SUKSES)) AND (nominal#4L > 0)) AND (nominal#4L < 50000000)) AND isnotnull(id_merchant#3))
## : +- InMemoryTableScan [waktu#0, id_trx#1, id_user#2, id_merchant#3, nominal#4L, status#5], [isnotnull(status#5), isnotnull(nominal#4L), (status#5 = SUKSES), (nominal#4L > 0), (nominal#4L < 50000000), isnotnull(id_merchant#3)]
## : +- InMemoryRelation [waktu#0, id_trx#1, id_user#2, id_merchant#3, nominal#4L, status#5], StorageLevel(disk, memory, deserialized, 1 replicas)
## : +- *(1) Scan ExistingRDD[waktu#0,id_trx#1,id_user#2,id_merchant#3,nominal#4L,status#5]
## +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, false]),false), [plan_id=73]
## +- Filter isnotnull(id_merchant#196)
## +- Scan ExistingRDD[id_merchant#196,kategori#197,kota#198]print("Jumlah baris hasil join:", joined.count())## Jumlah baris hasil join: 371658# (c) Total nominal per kota per kategori.
total_kk = (joined.groupBy("kota", "kategori")
.agg(F.sum("nominal").alias("total_nominal"))
.orderBy("kota", F.desc("total_nominal")))
print("C3(c) Total nominal per kota per kategori:")## C3(c) Total nominal per kota per kategori:total_kk.show(25, truncate=False)## +--------+------------+-------------+
## |kota |kategori |total_nominal|
## +--------+------------+-------------+
## |Bandung |Transportasi|1455560000 |
## |Bandung |Hiburan |1100050900 |
## |Bandung |Retail |1016883400 |
## |Bandung |Kuliner |898654300 |
## |Bandung |Kesehatan |668446100 |
## |Jakarta |Transportasi|1138838700 |
## |Jakarta |Hiburan |888514200 |
## |Jakarta |Kesehatan |778065700 |
## |Jakarta |Retail |556442400 |
## |Jakarta |Kuliner |214176200 |
## |Makassar|Retail |1344025800 |
## |Makassar|Kesehatan |1236338300 |
## |Makassar|Hiburan |1003119800 |
## |Makassar|Kuliner |580311300 |
## |Makassar|Transportasi|557000900 |
## |Medan |Transportasi|1439104900 |
## |Medan |Kesehatan |1008722600 |
## |Medan |Hiburan |1005036300 |
## |Medan |Kuliner |805227200 |
## |Medan |Retail |776654600 |
## |Surabaya|Kesehatan |1000164700 |
## |Surabaya|Hiburan |868802300 |
## |Surabaya|Transportasi|800425300 |
## |Surabaya|Kuliner |677970800 |
## |Surabaya|Retail |564119200 |
## +--------+------------+-------------+# (d) Window function: total kota, peringkat kategori, dan porsi (%).
w_kota = Window.partitionBy("kota")
w_rank = Window.partitionBy("kota").orderBy(F.desc("total_nominal"))
teratas = (total_kk
.withColumn("total_kota", F.sum("total_nominal").over(w_kota))
.withColumn("porsi_pct", F.round(100 * F.col("total_nominal") / F.col("total_kota"), 2))
.withColumn("peringkat", F.row_number().over(w_rank))
.filter(F.col("peringkat") == 1)
.select("kota", "kategori", "total_nominal", "porsi_pct")
.orderBy("kota"))
print("C3(d) Kategori teratas per kota (window function):")## C3(d) Kategori teratas per kota (window function):teratas.show(truncate=False)## +--------+------------+-------------+---------+
## |kota |kategori |total_nominal|porsi_pct|
## +--------+------------+-------------+---------+
## |Bandung |Transportasi|1455560000 |28.32 |
## |Jakarta |Transportasi|1138838700 |31.85 |
## |Makassar|Retail |1344025800 |28.47 |
## |Medan |Transportasi|1439104900 |28.58 |
## |Surabaya|Kesehatan |1000164700 |25.57 |
## +--------+------------+-------------+---------+# Versi Spark SQL (hasilnya harus sama).
total_kk.createOrReplaceTempView("total_kk")
print("C3(d) Versi Spark SQL:")## C3(d) Versi Spark SQL:spark.sql("""
SELECT kota, kategori, total_nominal,
ROUND(100 * total_nominal / total_kota, 2) AS porsi_pct
FROM (SELECT *,
SUM(total_nominal) OVER (PARTITION BY kota) AS total_kota,
ROW_NUMBER() OVER (PARTITION BY kota ORDER BY total_nominal DESC) AS rn
FROM total_kk)
WHERE rn = 1
ORDER BY kota
""").show(truncate=False)## +--------+------------+-------------+---------+
## |kota |kategori |total_nominal|porsi_pct|
## +--------+------------+-------------+---------+
## |Bandung |Transportasi|1455560000 |28.32 |
## |Jakarta |Transportasi|1138838700 |31.85 |
## |Makassar|Retail |1344025800 |28.47 |
## |Medan |Transportasi|1439104900 |28.58 |
## |Surabaya|Kesehatan |1000164700 |25.57 |
## +--------+------------+-------------+---------+Tafsiran C3. Dengan skema eksplisit,
waktu benar-benar bertipe timestamp dan
nominal bertipe long. Rencana eksekusinya
menampilkan BroadcastHashJoin, artinya tabel merchant yang
kecil (200 baris) dikirim ke semua executor sehingga 371.658
transaksi tidak perlu di-shuffle. Kategori teratas tiap kota:
Transportasi di Bandung (28,32%), Jakarta (31,85%), dan Medan (28,58%),
Retail di Makassar (28,47%), serta Kesehatan di Surabaya (25,57%). Hasil
window function dan Spark SQL sama persis.
4.4 Tugas C4: Ingestion dengan Structured Streaming (10 poin)
Bangun kueri streaming dari folder landing
(format JSON, skema eksplisit) dengan
trigger(availableNow=True),
maxFilesPerTrigger=1, checkpoint, dan
foreachBatch yang: (a) memisahkan event dengan
nominal null sebagai tidak valid, (b) menyimpan
event valid ke Parquet, dan (c) mencatat jumlah event
valid dan tidak valid per batch. Tampilkan catatan per
batch dan total baris Parquet.
from pyspark.sql.types import StructType, StructField, StringType, LongType
# Skema eksplisit: tidak ada inferensi tipe pada streaming (wajib untuk sumber berkas).
skema_ev = StructType([
StructField("id_trx", StringType()),
StructField("id_merchant", StringType()),
StructField("nominal", LongType()),
StructField("waktu", StringType()),
])
out_valid = os.path.join(folder, "out_valid") # tujuan Parquet event valid
ckpt = os.path.join(folder, "checkpoint") # lokasi checkpoint
catatan = [] # catatan per batch
def proses_batch(batch_df, batch_id):
"""Dijalankan untuk setiap micro-batch (satu berkas per batch)."""
valid = (batch_df.filter(F.col("nominal").isNotNull()) # (a) valid
.withColumn("waktu", F.to_timestamp("waktu")))
tidak_valid = batch_df.filter(F.col("nominal").isNull()) # (a) tidak valid
valid.write.mode("append").parquet(out_valid) # (b) simpan Parquet
catatan.append({"batch_id": batch_id, # (c) catatan batch
"valid": valid.count(),
"tidak_valid": tidak_valid.count()})
stream_df = (spark.readStream
.schema(skema_ev)
.option("maxFilesPerTrigger", 1) # satu berkas per batch
.json(landing))
kueri = (stream_df.writeStream
.foreachBatch(proses_batch)
.option("checkpointLocation", ckpt) # offset disimpan untuk pemulihan
.trigger(availableNow=True) # proses semua data lalu berhenti
.start())
kueri.awaitTermination()
print("C4 Catatan per batch:")## C4 Catatan per batch:for c in catatan:
print(" ", c)## {'batch_id': 0, 'valid': 486, 'tidak_valid': 14}
## {'batch_id': 1, 'valid': 479, 'tidak_valid': 21}
## {'batch_id': 2, 'valid': 482, 'tidak_valid': 18}print("Total event valid:", sum(c["valid"] for c in catatan),
"| tidak valid:", sum(c["tidak_valid"] for c in catatan))## Total event valid: 1447 | tidak valid: 53print("Total baris Parquet:", spark.read.parquet(out_valid).count())## Total baris Parquet: 1447Tafsiran C4. Karena
maxFilesPerTrigger=1, tiga berkas diproses sebagai tiga
batch (0, 1, 2) yang masing-masing berisi 500 event. Hasilnya 1.447
event valid disimpan ke Parquet dan 53 event (sekitar 3,5%) dipisahkan
karena nominal-nya null, sehingga 1.447 + 53 = 1.500.
Checkpoint menyimpan progres pemrosesan, jadi kalau kueri
dijalankan lagi, berkas yang sudah diproses tidak akan dibaca ulang dan
data tidak jadi dobel.
4.5 Tugas C5: EDA Skala Besar dan Insight (14 poin)
Gunakan DataFrame transaksi dari C3(a). (a) Buat laporan kualitas
data dalam satu select: jumlah nominal
negatif, nominal \(\ge\) 50 juta, dan
status PENDING. (b) Hitung persentil 5, 50, 95, dan 99
nominal transaksi valid dengan approxQuantile. (c) Hitung
jumlah transaksi SUKSES per jam (0–23) di Spark dan buat grafik
batangnya. (d) Tuliskan dua insight (temuan, bukti,
tindakan).
import matplotlib.pyplot as plt
# (a) Laporan kualitas data dalam SATU select (agregasi bersyarat).
print("C5(a) Laporan kualitas data:")## C5(a) Laporan kualitas data:(df_trx.select(
F.count("*").alias("total_baris"),
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_ge_50jt"),
F.sum(F.when(F.col("status") == "PENDING", 1).otherwise(0)).alias("status_pending"))
).show()## +-----------+---------------+---------------+--------------+
## |total_baris|nominal_negatif|nominal_ge_50jt|status_pending|
## +-----------+---------------+---------------+--------------+
## | 399920| 150| 80| 7884|
## +-----------+---------------+---------------+--------------+# (b) Persentil nominal valid (0 < nominal < 50 juta); galat relatif 0,001.
valid_all = df_trx.filter((F.col("nominal") > 0) & (F.col("nominal") < 50_000_000))
persentil = valid_all.approxQuantile("nominal", [0.05, 0.50, 0.95, 0.99], 0.001)
print("C5(b) Persentil nominal transaksi valid (Rp):")## C5(b) Persentil nominal transaksi valid (Rp):for nama, v in zip(["P5", "P50", "P95", "P99"], persentil):
print(" %-3s = %10s" % (nama, f"{v:,.0f}"))## P5 = 9,200
## P50 = 40,100
## P95 = 175,400
## P99 = 324,400# (c) Transaksi SUKSES per jam di Spark; hanya 24 baris hasil dibawa ke driver.
per_jam = (df_trx.filter(F.col("status") == "SUKSES")
.groupBy(F.hour("waktu").alias("jam"))
.count().orderBy("jam"))
pj = per_jam.toPandas()
print("\nC5(c) Rata-rata per jam:", round(pj["count"].mean()),
"| min:", pj["count"].min(), "| maks:", pj["count"].max())##
## C5(c) Rata-rata per jam: 15495 | min: 15246 | maks: 15780plt.figure(figsize=(9, 4))
plt.bar(pj["jam"], pj["count"], color="#2c7fb8")
plt.xticks(range(24))## ([<matplotlib.axis.XTick object at 0x0000014B02BBB8C0>, <matplotlib.axis.XTick object at 0x0000014B02CB4910>, <matplotlib.axis.XTick object at 0x0000014B02CB5090>, <matplotlib.axis.XTick object at 0x0000014B02CB5810>, <matplotlib.axis.XTick object at 0x0000014B02CB5F90>, <matplotlib.axis.XTick object at 0x0000014B02CB6710>, <matplotlib.axis.XTick object at 0x0000014B02CB6E90>, <matplotlib.axis.XTick object at 0x0000014B02CB7610>, <matplotlib.axis.XTick object at 0x0000014B02CB7D90>, <matplotlib.axis.XTick object at 0x0000014B02CD0550>, <matplotlib.axis.XTick object at 0x0000014B02CD0CD0>, <matplotlib.axis.XTick object at 0x0000014B02CD1450>, <matplotlib.axis.XTick object at 0x0000014B02CD1BD0>, <matplotlib.axis.XTick object at 0x0000014B02CD2350>, <matplotlib.axis.XTick object at 0x0000014B02CD2AD0>, <matplotlib.axis.XTick object at 0x0000014B02CD3250>, <matplotlib.axis.XTick object at 0x0000014B02CD39D0>, <matplotlib.axis.XTick object at 0x0000014B02CF4190>, <matplotlib.axis.XTick object at 0x0000014B02CF4910>, <matplotlib.axis.XTick object at 0x0000014B02CF5090>, <matplotlib.axis.XTick object at 0x0000014B02CF5810>, <matplotlib.axis.XTick object at 0x0000014B02CF5F90>, <matplotlib.axis.XTick object at 0x0000014B02CF6710>, <matplotlib.axis.XTick object at 0x0000014B02CF6E90>], [Text(0, 0, '0'), Text(1, 0, '1'), Text(2, 0, '2'), Text(3, 0, '3'), Text(4, 0, '4'), Text(5, 0, '5'), Text(6, 0, '6'), Text(7, 0, '7'), Text(8, 0, '8'), Text(9, 0, '9'), Text(10, 0, '10'), Text(11, 0, '11'), Text(12, 0, '12'), Text(13, 0, '13'), Text(14, 0, '14'), Text(15, 0, '15'), Text(16, 0, '16'), Text(17, 0, '17'), Text(18, 0, '18'), Text(19, 0, '19'), Text(20, 0, '20'), Text(21, 0, '21'), Text(22, 0, '22'), Text(23, 0, '23')])plt.xlabel("Jam (0-23)")
plt.ylabel("Jumlah transaksi SUKSES")
plt.title("Transaksi SUKSES per jam")
plt.tight_layout()
plt.show()
# (d) Bukti numerik untuk insight (semua dihitung di Spark/RDD, hasil kecil saja).
total_pos = df_trx.filter("nominal > 0").agg(F.sum("nominal")).first()[0]
total_ekstrem = df_trx.filter("nominal >= 50000000").agg(F.sum("nominal")).first()[0]
n_anom = df_trx.filter("nominal < 0 OR nominal >= 50000000").count()
n_total = df_trx.count()
print("\nBukti insight 1:")##
## Bukti insight 1:print(" Transaksi anomali:", n_anom, "(%.3f%% dari baris)" % (100 * n_anom / n_total))## Transaksi anomali: 230 (0.058% dari baris)print(" Total nominal positif: Rp", f"{total_pos:,}")## Total nominal positif: Rp 44,053,276,200print(" Nominal ekstrem: Rp", f"{total_ekstrem:,}",
"(%.1f%% dari total nominal positif)" % (100 * total_ekstrem / total_pos))## Nominal ekstrem: Rp 20,000,000,000 (45.4% dari total nominal positif)print(" Nominal ekstrem = %.0f x P99" % (250_000_000 / persentil[3]))## Nominal ekstrem = 771 x P99jml_gagal = sum(v for k, v in per_status if k == "GAGAL")
jml_pending = sum(v for k, v in per_status if k == "PENDING")
tot_trx = sum(v for _, v in per_status)
rata_gagal = 100 * jml_gagal / tot_trx
se = (rata_gagal / 100 * (1 - rata_gagal / 100) / 2000) ** 0.5 * 100
print("\nBukti insight 2:")##
## Bukti insight 2:print(" Tingkat kegagalan keseluruhan: %.2f%% | PENDING: %.2f%%" % (rata_gagal, 100 * jml_pending / tot_trx))## Tingkat kegagalan keseluruhan: 5.04% | PENDING: 1.97%print(" Merchant tertinggi %s: %.2f%% = %.1f galat baku di atas rata-rata (SE sekitar %.2f poin)"
% (top5[0][0], top5[0][3], (top5[0][3] - rata_gagal) / se, se))## Merchant tertinggi M164: 6.34% = 2.7 galat baku di atas rata-rata (SE sekitar 0.49 poin)Tafsiran C5(a)-(c). Ada 150 transaksi bernominal negatif (-1), 80 transaksi bernominal Rp50 juta ke atas, dan 7.884 transaksi PENDING (1,97%). Median nominal yang valid cuma Rp40.100 dan P99-nya Rp319.800, jadi angka Rp250 juta jelas jauh di luar kebiasaan. Untuk transaksi SUKSES per jam, jumlahnya merata sepanjang hari (sekitar 15.500 per jam) tanpa jam puncak yang jelas.
C5(d) Dua insight
Insight 1: sedikit data anomali bisa merusak total nominal.
- Temuan: ada 230 transaksi anomali (150 bernominal -1 dan 80 bernominal Rp250 juta). Jumlahnya cuma 0,058% dari seluruh baris, tapi dampaknya ke angka total besar sekali.
- Bukti: nominal ekstrem itu totalnya Rp20 miliar, atau 45,4% dari seluruh nominal positif (Rp44,05 miliar), dan nilainya sekitar 782 kali P99 (Rp319.800).
- Tindakan: pasang aturan validasi nominal saat data masuk (0 < nominal < batas wajar) dan pindahkan data yang tidak lolos ke area karantina, jangan langsung dibuang. Minta tim risiko memeriksa 80 transaksi Rp250 juta tersebut. Untuk laporan, pakai median atau persentil, bukan rata-rata. Pantau juga jumlah baris log yang rusak (80 baris, 0,02%).
Insight 2: selisih tingkat kegagalan antar-merchant kecil dan belum tentu berarti.
- Temuan: tingkat kegagalan secara keseluruhan 5,04%, sedangkan lima merchant teratas ada di 6,0-6,3% (M164 paling tinggi, 6,34%). Selain itu masih ada 1,97% transaksi berstatus PENDING.
- Bukti: tiap merchant punya sekitar 2.000 transaksi, sehingga galat bakunya sekitar 0,5 poin persentase, dan M164 hanya sekitar 2,7 galat baku di atas rata-rata. Dengan 200 merchant, selisih segitu masih bisa muncul secara kebetulan.
- Tindakan: jangan buru-buru menindak merchant tersebut. Pasang pemantauan dengan ambang batas dan lihat tren dari waktu ke waktu, cek kode error pada transaksi gagal di kelima merchant itu, dan rekonsiliasi transaksi PENDING dengan batas waktu otomatis. Karena beban transaksi merata sepanjang hari, kapasitas sistem tidak perlu disetel khusus untuk jam tertentu.