KAL
Menu

Implementasi

Tahap ingest, transform, serve.

Daftar isi

Desain pada topik sebelumnya kini dikerjakan sebagai satu skrip yang nyata. Data masukannya berupa berkas CSV kehadiran siswa yang sengaja berisi masalah umum, yaitu satu baris duplikat dan satu nilai kosong, dan keluarannya berupa tabel ringkasan per kelas pada lapisan gold. Skrip ini adalah penerapan paling langsung dari desain medallion, dan setiap tahapnya memakai topik yang sudah dipelajari pada modul-modul sebelumnya.

Data dan tiga tahap dalam satu skrip#

Aset proyek berupa satu berkas absen_mentah.csv berisi enam baris, dengan Budi bernilai kosong dan baris Ali muncul dua kali. Desain menuntut silver berisi satu baris valid per siswa dengan nilai terisi, sehingga pembersihan yang diharapkan jelas, yaitu membuang baris tanpa nilai dan membuang duplikat identitas.

python pipeline.py 
# tidak-dijalankan
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (
    SparkSession.builder
    .master("local[2]")
    .appName("pipeline-absen")
    .config("spark.sql.shuffle.partitions", "2")
    .getOrCreate()
)

akar = "/tmp/pipeline-absen"
import shutil
shutil.rmtree(akar, ignore_errors=True)

print("=== tahap ingest ===")
mentah = spark.read.option("header", "true").csv("absen_mentah.csv").cache()
print("baris mentah:", mentah.count())
mentah.coalesce(1).write.mode("overwrite").parquet(f"{akar}/bronze")

print("=== tahap transform ===")
bersih = (
    mentah
    .filter(F.col("nilai").isNotNull())
    .dropDuplicates(["id"])
    .withColumn("nilai", F.col("nilai").cast("int"))
    .withColumn("hari", F.lower(F.col("hari")))
)
print("baris bersih:", bersih.count())
bersih.coalesce(1).write.mode("overwrite").parquet(f"{akar}/silver")

print("=== tahap serve ===")
ringkas = (
    bersih
    .groupBy("kelas")
    .agg(F.count("*").alias("jumlah"), F.avg("nilai").alias("rata_rata"))
    .orderBy("kelas")
)
ringkas.coalesce(1).write.mode("overwrite").parquet(f"{akar}/gold")
gold = spark.read.parquet(f"{akar}/gold")
gold.show()

spark.stop()
text Ilustrasi keluaran 
WARNING: Using incubator modules: jdk.incubator.vector
=== tahap ingest ===
baris mentah: 6
=== tahap transform ===
baris bersih: 4
=== tahap serve ===
+-----+------+---------+
|kelas|jumlah|rata_rata|
+-----+------+---------+
|  10A|     2|     86.5|
|  10B|     2|     85.0|
+-----+------+---------+

Keluaran menunjukkan seluruh alur dalam beberapa angka. Enam baris mentah menjadi empat baris bersih, karena baris Budi tanpa nilai dibuang oleh filter dan duplikat Ali dilebur oleh dropDuplicates. Tabel gold menampilkan dua kelas dengan dua siswa per kelas, dan rata-rata 86.5 untuk 10A yang berasal dari Ali 85 dan Eko 88, serta 85.0 untuk 10B dari Citra 78 dan Dewi 92. Skrip ini menjalankan lima pekerjaan bertumpuk sehingga pada sandbox melewati batas verifikasi otomatis, dan keluarannya diambil dari pengujian manual yang skripnya identik, mengikuti pola modul streaming dan ekosistem.

Membaca keputusan di dalam skrip#

Setiap baris skrip menyimpan keputusan desain yang sudah dibahas sebelumnya, dan menelusurinya adalah cara belajar terbaik. .cache() pada pembacaan mentah memakai pola dari modul caching, karena DataFrame mentah dipakai oleh penghitungan baris, penulisan bronze, dan rantai transformasi, sehingga CSV tidak perlu diurai tiga kali. .coalesce(1) sebelum setiap penulisan menghasilkan satu berkas per lapisan pada data kecil, mengikuti pembahasan masalah berkas kecil pada modul berkas dan format. Rantai filter, dropDuplicates, dan withColumn adalah ringkasan seluruh modul transformasi, dan groupBy dengan agg menutup tahap serve dengan pola agregasi yang sama.

Pada proyek nyata, ketiga tahap sering berpindah menjadi tiga aplikasi terpisah yang disambung oleh lapisan penyimpanan, dan pada saat itu spark.read di awal setiap aplikasi menggantikan rantai DataFrame. Struktur folder dan makna tiap lapisan tetap sama persis, dan inilah alasan desain dipisahkan dari implementasi, karena keduanya berubah dengan kecepatan yang berbeda.

Memutar pipeline secara berkala#

Skrip di atas berjalan sekali dengan perintah python3, dan tingkat kematangan berikutnya adalah memutarnya berkala. Cara paling langsung adalah memanggil skrip yang sama dari penjadwal sistem, misalnya cron pada mesin yang sama, dengan berkas sumber diperbarui sebelum panggilan. Pola trigger availableNow pada modul streaming memberi jalan tengah yang lebih kuat, yaitu kueri yang memproses seluruh data yang tersedia lalu berhenti, sehingga penjadwal cukup menyalakan kueri pada jadwal tanpa mengurus berkas mana yang sudah diproses. Ketiga tingkat ini, yaitu sekali jalan, terjadwal, dan dipicu data, memakai skrip transformasi yang sama persis, dan perbedaannya hanya pada cara berkas masuk ke lapisan bronze.

Satu perubahan yang biasanya menyertai kenaikan tingkat adalah pindah dari Parquet ke tabel Delta pada ketiga lapisan, sebab tulis ulang berkala membutuhkan transaksi dan pembaca yang konsisten. Perpindahannya murah pada skrip ini, yaitu mengganti nilai format pada penulis dan menambah konfigurasi katalog yang sudah dijelaskan pada modul ekosistem. Struktur folder, tahapan, dan keluaran tidak berubah sama sekali.

Sebelum menutup implementasi, satu disiplin kecil perlu ditegakkan pada skrip, yaitu setiap tahap mencetak bukti barisnya sendiri. Dua angka pada keluaran tahap pertama dan kedua berfungsi sebagai pemeriksaan kesesuaian desain, dan ketika data sumber bertumbuh, selisih baris antar lapisan yang tiba-tiba berubah pola adalah sinyal paling awal bahwa sumber atau transformasi bergeser. Pemeriksaan berbentuk print sederhana ini menjadi dasar yang dipermakmur oleh pemantauan dan event log pada modul produksi, dan kebiasaan menempelkan bukti pada setiap tahap berpindah utuh ke pipeline yang jauh lebih besar.

Kesalahan umum#

Kesalahan pertama adalah menjalankan pipeline tanpa membersihkan lapisan lama. Mode overwrite menimpa isi lapisan yang ditulisnya, tetapi baris yang tersisa dari kegagalan sebelumnya tetap bisa mencemari pembacaan, dan shutil.rmtree pada awal skrip menjaga agar eksperimen berulang selalu mulai dari keadaan bersih. Pada produksi, kebersihan ini ditangani transaksi tabel Delta, bukan penghapusan folder.

Kesalahan kedua adalah menebak tipe pada pembacaan CSV dan membiarkan cast terjadi di tengah rantai. Skrip ini sengaja tetap membaca apa adanya lalu melakukan cast di silver, karena bronze memang menyimpan kebenaran mentah, tetapi pada sumber yang lebih besar schema eksplisit di pembacaan pertama lebih disiplin sebagaimana dibahas pada topik desain.

Kesalahan ketiga adalah menghitung ulang agregasi pada setiap kueri konsumen alih-alih menuliskannya ke gold. Tabel gold yang tertulis membuat konsumen membaca hasil jadi, dan pipeline yang gagal terlihat dari tabel yang tidak diperbarui barisnya, alih-alih tampilan yang lambat tanpa alarm.

Ringkasan#

  • Pipeline tiga tahap menyusun ingest apa adanya, transformasi pembersihan, dan agregasi yang ditulis ke gold, dengan data uji yang sengaja berisi duplikat dan nilai kosong.
  • Enam baris mentah menjadi empat baris bersih dan dua baris gold, dan setiap selisih baris dapat dijelaskan dari keputusan pembersihan yang eksplisit.
  • .cache(), .coalesce(1), dan rantai transformasi memakai pelajaran modul sebelumnya secara langsung, dan pemisahan tahap aplikasi hanya menggantikan rantai dengan pembacaan ulang lapisan.
  • Mode overwrite dengan reset folder menjaga pengulangan eksperimen, sementara produksi menitipkan kebersihan pada transaksi tabel Delta.

Referensi#