Post cover image
Photo by Jeff W on Unsplash

September 13, 2026

Laporan Praktikum Big Data :Apache Spark Installation & Introduction to PySpark DataFrames

Halo teman-teman semuanya perkenalkan nama saya Cahyo Adi Nugroho. Saya adalah salah satu mahasiswa teknologi informasi dari Universitas…

By Cahyoadi

14 min read

Halo teman-teman semuanya perkenalkan nama saya Cahyo Adi Nugroho. Saya adalah salah satu mahasiswa teknologi informasi dari Universitas Tidar. Di halaman medium ini saya akan menjelaskan laporan praktikum mata kuliah big data yang berfokus pada installasi apach spark dan pengenalan pyspark dateframe.

TUJUAN

  1. Menjelaskan konsep dasar Apache Spark dan perbedaannya dengan MapReduce klasik.
  2. Memasang dan mengonfigurasi Apache Spark beserta PySpark.
  3. Membuat SparkSession dan memuat data ke dalam Spark DataFrame dari disk lokal maupun HDFS.
  4. Melakukan operasi dasar DataFrame: select, filter, groupBy, agg.
  5. Membandingkan sintaks PySpark DataFrame dengan pandas DataFrame yang telah dipelajari di Pertemuan 2.

DASAR TEORI

Kenapa harus mencoba apache spark?

Apache Spark merupakan framework pemrosesan data terdistribusi yang dirancang untuk mengolah data berskala besar secara cepat dan efisien. Kemarin saya telah mempelajari Hadoop MapReduce sebagai mesin pemrosesan utama. Namun, MapReduce keterbatasan performa karena selalu menuliskan data hasil pemrosesan sementara ke dalam media penyimpanan fisik (disk HDFS) pada setiap tahap komputasi.

Berbeda dengan MapReduce, Apache Spark mengadopsi mekanisme pemrosesan di dalam memori. Komputasi dilakukan langsung di dalam RAM, sehingga pemrosesan data dapat berjalan lebih cepat dibandingkan MapReduce. Meskipun demikian, Spark tetap terintegrasi secara penuh dan dapat membaca maupun menulis data secara langsung dari lapisan penyimpanan HDFS.

Arsitektur utama apache spark

Komputasi terdistribusi pada Apache Spark terdiri dari tiga komponen utama dalam arsitekturnya:

  1. Driver Program: Proses utama yang menjalankan skrip program pengguna, mengoordinasikan seluruh alur komputasi, serta menyusun rencana eksekusi.
  2. Cluster Manager: Komponen yang bertugas mengalokasikan sumber daya komputasi di seluruh kluster. Cluster Manager dapat berjalan dalam mode local, Standalone, maupun memanfaatkan YARN pada kluster Hadoop.
  3. Executor: Proses pekerja yang berjalan pada node kluster untuk mengeksekusi tugas komputasi dan menyimpan data sementara di dalam memori.

Struktur data pada Spark

Dalam ekosistem Spark, terdapat dua struktur data utama:

  • RDD ( Resilient Distributed Dataset ): data dasar dan paling awal di Spark yang berupa kumpulan objek terdistribusi tanpa skema kolom yang jelas.
  • PySpark DataFrame: Abstraksi data terstruktur yang dibangun di atas RDD[5][6]. DataFrame memiliki bentuk tabel (terdiri atas baris dan kolom) yang mirip dengan pandas DataFrame atau tabel pada basis data relasional (SQL). PySpark DataFrame jauh lebih mudah digunakan dan dilengkapi dengan pengoptimalan eksekusi otomatis.

PRAKTIKUM

1. Download dan Ekstrak Apache Spark 3.5.9

Menginstall apache spark bisa dilakukan melalui terminal atau mengunjungi websitenya langsung di https://spark.apache.org/downloads.html . Disini saya melakukan installasi apache spark melalui terminal dengan mengetikkan perintah :

wget https://downloads.apache.org/spark/spark-3.5.9/spark-3.5.9-bin-hadoop3.tgz

tunggu proses download hingga selesai.

None

Seteleh selesai ekstrak dan pindah file yang sudah didownload tadi dengan mengetikan perintah dibawah ini satu persatu.

#perintah ekstrak
tar -xzvf spark-3.5.9-bin-hadoop3.tgz

#perintah memindahkan file
mv spark-3.5.9-bin-hadoop3 spark

Jika prosesnya berhasil maka saat saya melakukan ls ~/spark maka seharusya keluar file bin, sbin, python, dll

None

2. Konfigurasi Environment Variables

Pada pertemuan sebelumnya saya sudah mengkonfiguraso file bashrc untuk hadoop, dipertemuan kali ini saya akan mengkonfigurasi file bashrc untuk apache spark. Konfigurias ini bertujuan untuk menambahkan konfigurasi dan variabel lingkungan milik Apache Spark ke sistemm linux. Untuk melakukan konfigurasi tambahkan kode ini di bagian paling bawah file bashrc dengan cara mengetikkan perintah nano ~/.bashrc

# Konfigurasi Spark — Praktikum Big Data
export SPARK_HOME=$HOME/spark
export PATH=$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin
export PYSPARK_PYTHON=python3
None

Setelah itu simpan dan keluar menggunakan ctrl+o, enter, dan ctrl+x.

#untuk merefresh terminal setelah ditambahkan konfigurasi bashrc
source ~/.bashrc

#untuk menampikan okasi folder instalasi Apache Spark
echo $SPARK_HOME

#untuk memeriksa versi Apache Spark yang terinstal dan aktif saat ini
spark-submit --version

Jika proses konfigurasi berhasil maka outputnya seharusnya begini

None

3. Instalasi PySpark di Environment bigdata

Pysprak adalah library yag bertugas untuk menrejemahkan atau menghubungkan bahasa pemrograman pyhton dengan Apache spark. Karena itu saya harus menginstall pyspark terlebih dahulu. Untuk mengintall pyspark saya mengetikkan perintah ini diterminal

pip install pyspark==3.5.9
None

Setelah proses installasi selesai lakukan pengecekan dengan mengetikkan perintah

 python -c "import pyspark; print('pyspark', pyspark.__version__)"

Jika proses installasi berhasil maka outputnya seharusnya begini

None

4. Membuat SparkSession

SparkSession adalah kode python yang digunkan untuk menginisialisasi dan membuka sesi kerja utama Spark di dalam kode Python. SparkSession wajib dibuat di awal setiap script/notebook PySpark. Untuk menjalankan membuat sparksessio saya cukup menjalankan kode ini di cell jupyternotebook

from pyspark.sql import SparkSession

# Membuat SparkSession — "local[*]" berarti gunakan seluruh core CPU yang tersedia di VM
spark = SparkSession.builder \
    .appName("Pertemuan4-PengenalanPySpark") \
    .master("local[*]") \
    .getOrCreate()

# Mengurangi banyaknya pesan log teknis agar output lebih bersih
spark.sparkContext.setLogLevel("ERROR")

print("SparkSession berhasil dibuat!")
print("Versi Spark:", spark.version)
None

5. Memuat Data ke Spark DataFrame

Pada proses kali ini saya akan mencoba untuk memasukkan data bertipe ke CSV ke dalam struktur spark DateFrame.

  • Pembuatan data dummy

Pertama saya harus membuat data dummy untuk dimasukkan ke spark dataframe. Untuk membuat data dummynya saya menggunakan kode pyhton pintar yang akan embuat file .CSV pada pertemuan kedua yang telah disediakan di modul. Kode ini akan mengecek apakah file .CSV pertemuan kedua tersebut masih ada di penyimpanan laptopku atau tidak, kalo masih ada maka data dummy tidak dibuat, namun jika tidak maka data dummy di buat.

# Membuat ulang dataset contoh (identik dengan Tugas Mandiri Pertemuan 2) jika belum ada
import os
if not os.path.exists("data_transaksi_ecommerce.csv"):
    import numpy as np
    import pandas as pd
    np.random.seed(42)
    n = 600
    kategori_list = ["Elektronik", "Fashion", "Makanan & Minuman", "Kesehatan & Kecantikan", "Rumah Tangga"]
    kota_list = ["Magelang", "Yogyakarta", "Semarang", "Solo", "Purworejo"]
    metode_bayar_list = ["Transfer Bank", "E-Wallet", "COD", "Kartu Kredit"]
    tanggal_range = pd.date_range("2026-07-01", "2026-07-31", freq="D")
    data = {
        "order_id": [f"ORD-{1000+i}" for i in range(n)],
        "tanggal": np.random.choice(tanggal_range, size=n).astype(str),
        "kategori": np.random.choice(kategori_list, size=n, p=[0.25,0.25,0.20,0.15,0.15]),
        "kota": np.random.choice(kota_list, size=n),
        "unit_terjual": np.random.randint(1, 10, size=n),
        "harga_satuan": np.random.choice([25000,50000,75000,100000,150000,250000,500000], size=n),
        "metode_pembayaran": np.random.choice(metode_bayar_list, size=n, p=[0.35,0.30,0.20,0.15]),
    }
    pd.DataFrame(data).to_csv("data_transaksi_ecommerce.csv", index=False)
    print("Dataset dibuat ulang.")
else:
    print("Dataset sudah tersedia.")

Import pustaka os berfungsi untuk berinteraksi dengan sistem operasi tempat program berjalan yang nantinya digunakan melakukan pengecekan keberadaan berkas data_transaksi_ecommerce.csv di direktori kerja. Penggunaan np.random.seed(42) memastikan data dummy yang dihasilkan selalu konsisten (reprodusibel).

  • Memasukkan file CSV kedalam spark dataframe

Di cell kedua atau tahap kedua ini berfungsi untuk membaca file CSV dan mengonversinya menjadi objek Spark DataFrame

# Membaca berkas CSV lokal menjadi Spark DataFrame
# header=True    -> baris pertama dianggap nama kolom
# inferSchema=True -> Spark otomatis menebak tipe data tiap kolom (angka, teks, dst.)
df = spark.read.csv("data_transaksi_ecommerce.csv", header=True, inferSchema=True)

print("Tipe objek:", type(df))
df.printSchema()

kode ini Memanggil metode spark.read.csv() milik SparkSession untuk membaca berkas data_transaksi_ecommerce.csv. Selain itu, Parameter header=Truememberi tahu Spark bahwa baris pertama pada file CSV berisi nama-nama header/kolom, bukan baris data. ParameterinferSchema=Truememerintahkan Spark untuk menganalisis isi data dan secara otomatis menentukan tipe data terbaik untuk setiap kolom (misalnya: kolom angka dibaca sebagaiInteger, tanggal/teks dibaca sebagai String). Objek hasil pembacaan ini disimpan dalam variabel dfberbentuk Spark DataFrame yang siap diproses terdistribusi.df.printSchema(): Menampilkan struktur skema tabel, nama seluruh kolom, serta tipe data hasil inferensi (misal: string, integer, double`) beserta informasi apakah kolom tersebut boleh bernilai kosong atau tidak.

  • Menampilkan tabel data sampel

Cell ini berfungsi untuk melakukan verifikasi atau pengcekean terhadap DataFrame yang berhasil dimuat.

# Menampilkan beberapa baris pertama — mirip df.head() di pandas, namun disebut show()
df.show(5)

# Menghitung jumlah baris — mirip len(df) di pandas
print("Jumlah baris:", df.count())

df.show(5): Menampilkan 5 baris pertama data dalam format tabel terstruktur di terminal/Jupyter. Ini digunakan untuk memverifikasi bahwa data CSV telah terbaca dengan rapi ke dalam kolom yang sesuai.

Spark DataFrame berbeda dengan Pandas DataFrame. Walaupun pemanggilannya mirip, Spark DataFrame bekerja secara lazy evaluation (pemrosesan baru benar-benar dijalankan saat dipanggil aksi seperti .show()) sehingga jauh lebih cepat dan efisien untuk menangani Big Data bertipe terdistribusi.

6. Operasi Dasar DataFrame: Select, Filter, GroupBy, Agg

Kali ini saya akan memcoba beberapa fungsi-fungsi dasar utama untuk memanipulasi, memfilter, mengubah, dan mengagregasi data di pyspark.

  • Memilih kolom tertentu

fungsi select adalah untuk Memilih kolom-kolom tertentu saja dari DataFrame yang ingin ditampilkan atau diolah. Sama seperti pada SQL select digunakan untuk menyaring kolom agar tampilan/proses data lebih fokus dan tidak memuat seluruh kolom yang ada.

None

Select(

  • Menyaring Baris

Untuk menyaring baris, bisa menggunakan fungsi filter(). Funsi filter() akan Menyaring baris data berdasarkan kriteria atau kondisiyang telah ditentukan. Mirip seperti where pada SQL, yang tujuannya untuk mengambil baris yang memenuhi syarat.

None
  • Mengelompokkan dan Mengagregasi Data

Sama seperti pada SQL group by digunakan untuk mengelompokkan baris berdasarkan kategori yang telah ditentukan.

from pyspark.sql.functions import sum as spark_sum, count, avg

# Menambahkan kolom baru: total_pendapatan = unit_terjual x harga_satuan
df = df.withColumn("total_pendapatan", col("unit_terjual") * col("harga_satuan"))

# Meringkas: total pendapatan & jumlah transaksi per kota, diurutkan dari tertinggi
ringkasan_kota = df.groupBy("kota").agg(
    spark_sum("total_pendapatan").alias("total_pendapatan"),
    count("order_id").alias("jumlah_transaksi"),
    avg("unit_terjual").alias("rata_rata_unit")
).orderBy(col("total_pendapatan").desc())

ringkasan_kota.show()

Pada kode diatas ada beberapa fungsi agregasi serta groupby,

— > df.withColumn("total_pendapatan", df["unit_terjual"] * df["harga_satuan"]) akan Menambahkan kolom baru bernama total_pendapatan yang nilainya didapat dari perkalian antara perkalian jumlah unit dan harga satuan.

→ df.groupBy("kategori").sum("unit_terjual") adalah fungsi goupby. Fungsi ini akan menghitung total penjualan, rata-rata pendapatan, atau jumlah transaksi per kategori produk atau per kota.

→ df.orderBy(df["total_pendapatan"].desc())adalah fungsi yang akan mengurutkan data. fungsi ini akan menampilkan data dari yang tertinggi ke terkecil (atau sebaliknya), misalnya mencari produk paling laku atau kota dengan pendapatan terbesar.

None

Dengan konsep pengelompokan (group by) akan membuat ringkasan atau hasil yang mudah untuk dianalisis. Output diatas membuat saya lebih menganalisis kota mana yan menghasilkan pendapatan paling banyak. Hal ini sangat berguna dalam case dunia nyata

7. Membaca Data Langsung dari HsayDFS

Kali ini saya akan mengintrgrasikan HDFS dengan sprark. Hadoop disini sebagai penyimpanannya dan spark sebagai pemrosesnya. Ini juga membuktkan bahwa spark tidak hanya membaca dari disk lokal, tetapi juga mampu mengolah data hdfs secara langsung dan tanpa harus mengunduhnya terlebih dahulu ke lokal.

# Upload dataset ke HDFS
!hdfs dfs -mkdir -p /user/mahasiswa/pertemuan4
!hdfs dfs -put -f data_transaksi_ecommerce.csv /user/mahasiswa/pertemuan4/

# Membaca CSV LANGSUNG dari HDFS menggunakan Spark
df_dari_hdfs = spark.read.csv(
    "hdfs://localhost:9000/user/mahasiswa/pertemuan4/data_transaksi_ecommerce.csv",
    header=True, inferSchema=True
)
print("Jumlah baris dari HDFS:", df_dari_hdfs.count())
df_dari_hdfs.show(5)
  • Pertama dalam kode tersebut adalah membuat direktori mkdir dan mengupload file .csv ke hdfs menggunakan -put.
  • spark.read.csv() adalah metode bawaan PySpark untuk membaca berkas berformat CSV.
  • url 'hdfs://localhost:9000/user….' maksdunya adalah Alamat URI jaringan terdistribusi HDFS alamat jalur terdistribusi yang memberi tahu PySpark ke mana harus mencari dan mengambil data di dalam cluster Hadoop.
  • header=True dan inferSchema=True berfungsi agar baris paling atas adalah nama kolom dan mendeteksi tipe data tiap kolom secara otomatis.
  • fungsi 'df_dari_hdfs.count()' adalah perintah hfds yang berfungsi untuk menghitung total baris pada dara yang berhasil dimuat.
None

LATIHAN MANDIRI

Latihan mandiri ini saya akan mencoba menerapkan perinttah pyspark untuk membuat hasil atau kesimpulan yang diminta oleh soal.

Membuat data (persiapan)

# Jalankan ini dulu sebelum mengerjakan latihan di bawah
spark = SparkSession.builder.appName("Latihan4").master("local[*]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
df = spark.read.csv("data_transaksi_ecommerce.csv", header=True, inferSchema=True)
df = df.withColumn("total_pendapatan", col("unit_terjual") * col("harga_satuan"))
print("Siap. Jumlah baris:", df.count())

Kode tersebut akan membuat fie data_transaksi_ecommerce.csv yang berbeda dari file yang digunakan pada praktikum. Kode tersebut juga menambahkan kolom total_pendapatan yang nantinya digunakan pada soal latihan. Pada intinya kode memuat ulang dataset asli data_transaksi_ecommerce.csv dari awal.

None

Dan beginilah penampakan data dari file .csv yang barusan saya buat.

Soal 1 (menampilkan order_id, kategori, dan total_pendapatan dari kategori metode pembayaran e-wallet)

# Jawaban Soal 1 di sini

from pyspark.sql.functions import col

df.filter(col("metode_pembayaran") == "E-Wallet").select("order_id", "kategori", "total_pendapatan").show()
  • Pertama saya harus mengimpor fungsi col (column) dari pustaka pyspark.sql.functions . pengimporan ini bertujuan untuk memungkinkan PySpark merujuk dan memanipulasi kolom data secara eksplisit sebagai objek kolom.
  • df.filter(col("metode_pembayaran") == "E-Wallet") berfungsi menyaring baris data yang memiliki nilai 'E-Walleyt' pada kolom metode_pembayaran
  • .select("order_id", "kategori", "total_pendapatan") membatasi output agar yang ditampilkan hanya kolom order_id, kategori, dan total_pendapatan sesuai instruksi soal.
None

Soal 2 (mwnghitung total pendapatan per kategori dan diurutkan dari yang paling tinggi)

# Jawaban Soal 2 di sini

from pyspark.sql.functions import sum, col

df.groupBy("kategori").agg(sum("total_pendapatan").alias("total_pendapatan")).orderBy(col("total_pendapatan").desc()).show()
  • Langkah pertama adalah mengmpor fungsi agregasi sum yang digunakan untuk menjumlahkan nilai-nilai numerik pada suati kolom dan col untuk merujuk pada kolom tertentu dari pustakan pyspark.sql.functions
  • df.groupBy("kategori") ini adalah kode inti yang mengelompokan baris berdasarkan kolom kategori.
  • .agg(sum("total_pendapatan").alias("total_pendapatan")) kode ini akan menjumlahkan nilai pada total pendapatan berdasarkan kelompok kategori masing-masing.
  • .orderBy(col("total_pendapatan").desc()) kode ini akan mengurutkan nilai pada kolom total_pendapatan dari yang terbesar hingga terkecil.
None

Soal 3 (menampilkan jumlah transaksi untuk masing-masing metode_pembayaran (tidak pakai agg)

# Jawaban Soal 3 di sini

df.groupBy("metode_pembayaran").count().show()
  • Pada kode ini saya tidak perlu mengimpoir fungsi apapun karena kodenya cukup simpel
  • groupBy("metode_pembayaran") akan mengelompokkan baris berdasarkan janis metode pembayarannya.
  • .count() adalah fungsi agregasi ringkas PySpark untuk menghitung banyaknya baris/transaksi dalam setiap kelompok secara otomatis tanpa memerlukan fungsi .agg().
None

Soal 4 Refleksi Dalam 2–3 kalimat: apa yang dimaksud dengan lazy evaluation di PySpark, dan mengapa hal ini menguntungkan ketika bekerja dengan data berskala besar?

  • Jawaban :

Lazy evaluation adalah sifat PySpark di mana operasi transformasi data (seperti filter() atau select()) tidak langsung dieksekusi saat kode ditulis, melainkan hanya menyusun rencana eksekusinya terlebih dahulu. Proses komputasi baru benar-benar dijalankan ketika memanggil perintah action seperti .show(), .count(), atau .collect(). Sifat ini sangat menguntungkan untuk data berskala besar karena memungkinkan Spark mengoptimalkan keseluruhan rencana eksekusi secara otomatis sebelum benar-benar memproses data.

TUGAS MANDIRI

None

Membuat datasest baru

# Sel ini membuat dataset baru untuk Tugas Mandiri Pertemuan 4 dan mengunggahnya ke HDFS
import numpy as np
import pandas as pd

np.random.seed(99)
n = 1000
kategori_list = ["Elektronik", "Fashion", "Makanan & Minuman", "Kesehatan & Kecantikan", "Rumah Tangga", "Olahraga"]
kota_list = ["Magelang", "Yogyakarta", "Semarang", "Solo", "Purworejo", "Kebumen"]
metode_bayar_list = ["Transfer Bank", "E-Wallet", "COD", "Kartu Kredit"]
tanggal_range = pd.date_range("2026-09-01", "2026-09-30", freq="D")

data = {
    "order_id": [f"ORD-{3000 + i}" for i in range(n)],
    "tanggal": np.random.choice(tanggal_range, size=n).astype(str),
    "kategori": np.random.choice(kategori_list, size=n),
    "kota": np.random.choice(kota_list, size=n),
    "unit_terjual": np.random.randint(1, 12, size=n),
    "harga_satuan": np.random.choice([20000, 45000, 60000, 90000, 125000, 200000, 350000], size=n),
    "metode_pembayaran": np.random.choice(metode_bayar_list, size=n),
    "rating": np.random.choice([1, 2, 3, 4, 5, np.nan], size=n, p=[0.03, 0.02, 0.10, 0.30, 0.35, 0.20]),
}
df_tugas4 = pd.DataFrame(data)
df_tugas4.to_csv("transaksi_september_2026.csv", index=False)
print(f"Dataset dibuat: {df_tugas4.shape[0]} baris")

# Mengunggah ke HDFS
!hdfs dfs -mkdir -p /user/mahasiswa/tugas4
!hdfs dfs -put -f transaksi_september_2026.csv /user/mahasiswa/tugas4/
print("Berhasil diunggah ke HDFS: /user/mahasiswa/tugas4/transaksi_september_2026.csv")

kode ini akan membuat dataset penjualan ecommerce berjumlah 1000 baris lalu merubahnya menjadi dataframe lalu disimpan di disk lokal sebagai transaksi_september_2026.csv . Selain itu, kode tersebut juga akan mengunggah file .csv tadike hdfs dengan -put. yang nantiya akan digunkan menggunakan pyspark pada soal selanjutnya

None

Memulai pyspark session

None

Pada tugas tertera bahwa pengerjaanya harus di notebook baru, karena hal ini saya juga harus memulai sparksession baru. Alasan mengapa saya harus memulai sparksession baru adalah karena variabel dan koneksi Spark dari notebook sebelumnya tidak akan terbawa otomatis ke notebook baru.

#memulai sparksession di notebook baru

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("TugasMandiri4") \
    .master("local[*]") \
    .getOrCreate()

spark.sparkContext.setLogLevel("ERROR")

pengimporan kelas sparksession dair pustaka pyspark.sql bertujuan untuk membuat koneksi ke mesin spark. variabel spark akan menyimpan semua objek yang berhasil dibuat pada sesi ini. Mesin spark dijalankan pada mode lokal yang artinya dijlankan pada laptop dan diberi nama TugasMandiri4. .getOrCreate() adalah fungsi yang akan mengambil SparkSession yang sudah aktif jika ada, atau membuat SparkSession baru jika belum ada. Ini mencegah terjadinya bentrok akibat pembuatan banyak sesi Spark dalam satu waktu.

None

A. Membaca dan Eksplorasi Awal

#A. membaca dan eksplorasi data

#menentukan lokasi berjas data di dalaam hdfs
path_hdfs = "hdfs://localhost:9000/user/mahasiswa/tugas4/transaksi_september_2026.csv"
df = spark.read.csv(path_hdfs, header=True, inferSchema=True)

#menampilkan skema
df.printSchema()

#menampilkan 10 baris pertama
df.show(10)

#menampilkan total baris
print("Total baris : ", df.count())
  • path hfds : adalah variabel yang menyimpan alamat URL tempat berks csv berada dalam hdfs. Singktanya pencantuman URL adalah cara memberitahu kepada mesin spark dimana file yang akan diproses disimpan dalam hdfs.
  • df = spark.read.csv(path_hdfs, header=True, inferSchema=True) : adalah metode dari sparksession yang membaca berksa berformat csv. Sleain itu header=True memberitahu spark bahwa baris pertama dalam csv adalah nama kolom bukan data dan inferSchema:True adalah perintah yang otomatis mendeteksi tipe data dari setiap kolom.
  • df.printSchema() : fungsi menampilkan skema dataframe pada output.
  • print("Jumlah baris:", df.count()) : akan menghitung total seluruh baris dari data hdfs.
  • df.show(10) : menampilkan data pada 10 baris pertama
None

B. Menangani nilai NULL

#B. Menangani Data Kosong
from pyspark.sql.functions import col

#menghitung jumlah baris yang memiliki nilai kosong pada kolom rating
jumlah_null = df.filter(col("rating").isNull()).count()
print(f"jumlah baris kosong pada kolom (NULL) pada kolom rating: {jumlah_null}")

#memilih na.fill() untuk penanganan nilai null pada kolom rating
penanganan = df.na.fill({"rating":0})

#memastikan bahwa nilai null sudah tidak ada lagi
jumlah_null2 = penanganan.filter(col("rating").isNull()).count()
print(f"jumlah null pada kolom rating setelah ditangani: {jumlah_null2}")
  • kode diatas harus mengimpor fungsi col untuk merujuk pada kolom rating secara eksplisit
  • variabel jumlah_null akan akan menyimpan penyaringan baris yang memiliki nilai null pada kolom rating menggunakan konsep filter dan menghitungnya.
  • variabel penanganan digunakan untuk menggantikan nilai null menjadi 0
  • vairabel jumlah_null2 adalah digunkan untuk memastikan bahwa fungsi dari variabel penanganan berhasil.
None
  • ALASAN SAYA MEMILIH na.fll()* na.fill() akan menggantikan nilai null menjadi 0 sedangkan na.drop() akan menghapus baris yang memiliki nilai null pada kolom rating. Jika saya memilih na.drop() maka seluruh data yang memiliki rating null akan dihapus dan ini akan menyebabkan analisis proses bisnis menjadi terganggu karena jumlah data yang memiliki nilai null pada kolom rating sebanyak 204 atau 20% lebih.

C. Transformasi Data

#C. Transfromasi data
from pyspark.sql.functions import col, when

#menambahkan kolom total_pendapatan
transformasi = penanganan.withColumn("total_pendapatan", col("unit_terjual") * col("harga_satuan"))

#menambahkakn kolom tier_transaksi
transformasi = transformasi.withColumn("tier_transaksi",when(col("total_pendapatan") > 500000, "besar").otherwise("kecil"))

#menampilkan 10 baris pertam
transformasi.show(10)
  • kode kali ini mengimpor fungsi col untuk mengambil variabel kolom dan when untuk percabangan
  • variabel transformasi akan menyimpan perubuahan yaitu menambahkan kolom total pendapatan yang dari hasil perkalian dari unit_terjual dan harga_satuan
  • Setelah kolom total_pendapatan tersimpan, variabel transformasi akan menyimpan perubahan yang saya buat berupa kolom tier_transaksi. Kolom tier_transaksi akan mengevaluasi apakah nilai total_pendapatan lebih dari 500000 atau tidak. Jika total_pendapatan suatu baris diatas 500000 maka baris tersebut dilabeli besar pada kolom tier_transaksinya, jika tidak maka dilabeli kecil.
None

D. Analisis dengan GroupBy

#D. Analisis dengan GroupBy
from pyspark.sql.functions import sum as spark_sum, count, avg, col

#kategori dengan total pendapatan paling tinggi
print("1. Kategori dengan total pendapatan tertinggi")
transformasi.groupBy("kategori").agg(spark_sum("total_pendapatan").alias("total_pendapatan")).orderBy(col("total_pendapatan").desc()).show(1)

#kota dengan jumlah tier transaksi 'besar' terbanyak
print("2. Kota dengan tier tansaksi 'besar' terbanyak")
transformasi.filter(col("tier_transaksi") == "Besar").groupBy("kota").agg(count("order_id").alias("transaksi_besar")).orderBy(col("transaksi_besar").desc()).show(1)

#rata_rata rating untuk masing-masing metode pemayaran
print("3. rata-rata ratig per metode pembayaran")
transformasi.groupBy("metode_pembayaran").agg(avg("rating").alias("rata_rata_rating")).show()
  • kode ini mengimpor banyak sekali fungsi karena pada soal d akan dibutuhkan beberapa operasi. fungsi sum diubah namanya menjadi spark_sum untuk menghindari konflik dengan fungsi sum() bawaan Python.
  • soal pertama : groupBy("kategori") akan menghitung total pendapatan menggunakan spark_sum dan mengurutkannya secara descending. saya hanya mengambil data teratas karena soal meminta kategori dengan pendapatan tertinggi.
  • soal kedua : konsep filter akan menyairng baris data yang bernilai Besar, lalu dikelompokkan berdasarkan kota dan dihitung jumlah transaksinya menggunakan count().
  • soal ketiga : groupBy("metode_pembayaran") akan mengelompokkan data berdasarkan metode_pembayaran lalu menghitung rata-rata pada kolom rating yang sudah dibreishkan pada soal B.
None

E. Menyimpan Hasil ke HDFS

#E. Menyimpan Hasil ke HDFS

#Menentukan path tujuan di HDFS
path_simpan_hdfs = "hdfs://localhost:9000/user/mahasiswa/tugas4/hasil_transformasi_csv"

#Menyimpan DataFrame hasil transformasi C ke HDFS dalam format CSV
transformasi.write.mode("overwrite").csv(path_simpan_hdfs, header=True)
print("Berhasil menyimpan data ke HDFS!")

# Verifikasi: Membaca kembali data yang baru disimpan dari HDFS
df_verifikasi = spark.read.csv(path_simpan_hdfs, header=True, inferSchema=True)
print("Verifikasi Berhasil! Jumlah baris data yang dibaca kembali:", df_verifikasi.count())
df_verifikasi.show(5)
  • pertama saya harus menentukan path tujuan di hdfs dengan mencatumkan url path. Setelah itu saya harus menyimpan datafrmae hasil transformasi pada poin c ke hdfs menggunakan frmat csv, pada tahap ini dilengkapi dengan kode mode("overwrite") yang artinya jika file hasil_transformasi_csv sudah ada, Spark akan menimpanya (overwrite) sehingga tidak terjadi error Folder Already Exists. Terakhir pada proses verifikasi, spark membaca ulang folder hasil simpanan di HDFS untuk membuktikan bahwa data tidak rusak dan seluruh barisnya berhasil tersimpan sempurna.
None

Mengapa Hasil Simpanan Spark Terdiri dari Beberapa Berkas Partisi (part-00000…)?

  • jawaban :

Spark menyimpan hasil olahan data ke dalam bentuk beberapa berkas partisi (part-00000-….csv, part-00001-….csv, dst.) dan bukan satu berkas tunggal karena Spark bekerja menggunakan arsitektur komputasi terdistribusi atau paralel. Data dibagi ke dalam beberapa blok yang diproses secara independen oleh Core CPU/Executor yang berbeda. Ketika proses penyimpanan dilakukan, setiap Executor menuliskan bagian datanya masing-masing secara bersamaan (paralel) langsung ke HDFS untuk mengoptimalkan kecepatan write (throughput). Hal ini berbeda dengan pustaka single-threaded seperti pandas yang menulis seluruh data ke dalam satu berkas tunggal secara sekuensial.

KESIMPULAN

Dari seluruh rangkaian praktikum yang telah saya lakukan, dapat disimpulkan bahwa alur analisis data transaksi yang sebelumnya berbasis pandas berhasil direplikasi dengan baik menggunakan PySpark DataFrame. Pengolahan data berskala besar menjadi jauh lebih cepat dan efisien berkat dukungan pemrosesan di dalam memori serta sifat lazy evaluation PySpark yang otomatis mengoptimalkan alur komputasi sebelum eksekusi dijalankan. Selain itu, integrasi antara HDFS sebagai media penyimpanan dan PySpark sebagai mesin pemroses terbukti berjalan lancar, baik saat membaca data transaksi secara langsung maupun ketika menyimpan kembali hasil olahannya ke dalam bentuk berkas partisi terdistribusi.

Penulis : Cahyo Adi Nugroho, Mahasiswa S1 Teknologi Informasi,Fakultas Teknik, Universitas Tidar Magelang

Referensi : Modul_Praktikum_BigData_Pertemuan4.ipynb

Tautan Github : https://github.com/cahyoadi-hash/Practic-Big-data/blob/9de9454c87cc6fe768c6c6ce4a8a9b7c0cc809bb/TUGAS4_2505060034_Cahyo%20Adi%20Nugroho.ipynb