Belajar Apache Spark - Machine Learning & MLlib

#Pendahuluan

Apache Spark bukan hanya engine ETL dan SQL — di dalamnya ada MLlib, library machine learning yang scalable. Dengan MLlib, kalian bisa melatih model klasifikasi dan regresi di atas dataset berukuran terabyte tanpa harus memindahkan data ke sistem lain.

Mengapa MLlib penting? Sebagian besar pipeline ML di produksi diawali oleh Spark: data dibersihkan dan disiapkan di Spark, model dilatih dengan MLlib, lalu hasilnya dipakai untuk prediksi batch atau real-time. Memahami MLlib berarti memahami bagaimana machine learning dijalankan dalam skala enterprise.

Episode ini membahas empat hal: konsep pipeline API dengan transformers dan estimators, feature engineering, pelatihan dan tuning model, serta use cases klasifikasi, regresi, clustering, dan recommendation.

#Pengenalan MLlib dan Pipeline API

#Konsep Pipeline

Pipeline di MLlib mengikuti pola yang mirip dengan sklearn: serangkaian stages yang dijalankan berurutan. Ada dua jenis stage:

• Transformer: mengubah DataFrame, biasanya menambah kolom — contoh VectorAssembler, StringIndexer.
• Estimator: dipelajari dari data (fit) dan menghasilkan model yang kemudian menjadi transformer — contoh LogisticRegression, RandomForestClassifier.

data → StringIndexer → VectorAssembler → LogisticRegression → prediksi
       (transform)       (transform)       (estimator → model)

Kelebihan pipeline: seluruh alur fit dan transform dibungkus sebagai satu objek, sehingga bisa disimpan, dimuat, dan dijalankan ulang dengan konsisten.

#Membangun Pipeline

Pipeline sederhana

from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.feature import VectorAssembler, StringIndexer
 
indexer = StringIndexer(inputCol="kategori", outputCol="kategori_idx")
assembler = VectorAssembler(
    inputCols=["umur", "jumlah", "kategori_idx"],
    outputCol="features")
lr = LogisticRegression(featuresCol="features", labelCol="churn")
 
pipeline = Pipeline(stages=[indexer, assembler, lr])

Pipeline(stages=[...]) mendefinisikan alur. Pada tahap ini belum ada komputasi — model baru dipelajari saat pipeline.fit() dipanggil.

#Feature Engineering: Transformers dan Estimators

#Transformers Dasar

Feature engineering adalah proses mengubah kolom mentah menjadi fitur yang dipakai model. Transformers paling umum:

• StringIndexer: mengubah kategori string menjadi indeks numerik.
• OneHotEncoder: mengubah indeks kategori menjadi representasi biner.
• VectorAssembler: menggabungkan beberapa kolom menjadi satu vektor fitur.
• StandardScaler: menormalisasi skala fitur numerik.

Feature engineering lengkap

from pyspark.ml.feature import OneHotEncoder, StandardScaler
 
encoder = OneHotEncoder(inputCol="kategori_idx", outputCol="kategori_vec")
scaler = StandardScaler(inputCol="features_raw", outputCol="features")
 
pipeline = Pipeline(stages=[indexer, encoder, assembler_raw, scaler, lr])

StandardScaler penting saat fitur punya rentang sangat berbeda — misalnya umur di angka puluhan dan jumlah transaksi di jutaan. Tanpa scaling, model seperti regresi akan didominasi fitur dengan skala terbesar.

#Estimator dan Model

Setelah fit, estimator menghasilkan model yang bisa dipakai untuk transformasi. Model-model klasik yang tersedia di MLlib antara lain:

• LogisticRegression: klasifikasi biner dan multinomial.
• RandomForestClassifier dan GBTClassifier: tree ensemble.
• LinearRegression: regresi.
• KMeans: clustering.
• ALS: collaborative filtering untuk recommendation.

#Model Training, Tuning, dan Evaluasi

#Melatih dan Mengevaluasi Model

Latih pipeline lalu evaluasi dengan metrik:

Fit, predict, dan evaluasi

model = pipeline.fit(data_latih)
prediksi = model.transform(data_uji)
 
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
evaluator = MulticlassClassificationEvaluator(labelCol="churn", metricName="accuracy")
print("Akurasi:", evaluator.evaluate(prediksi))

model.transform(data_uji) menjalankan seluruh pipeline — indexing, encoding, scaling, hingga prediksi — dalam satu panggilan. Evaluator mengukur kinerja model pada data yang belum pernah dilihat.

#Tuning dengan CrossValidator

Mencari hyperparameter terbaik sebaiknya dilakukan secara sistematis dengan CrossValidator atau TrainValidationSplit:

Tuning dengan CrossValidator

from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
 
grid = ParamGridBuilder() \
    .addGrid(lr.regParam, [0.01, 0.1]) \
    .addGrid(lr.maxIter, [10, 50]) \
    .build()
 
cv = CrossValidator(
    estimator=pipeline,
    estimatorParamMaps=grid,
    evaluator=evaluator,
    numFolds=3)
 
cv_model = cv.fit(data_latih)

cv.fit(data_latih) melatih setiap kombinasi hyperparameter pada beberapa fold dan memilih yang terbaik berdasarkan evaluator. Perlu diingat: tuning berulang kali melatih model, jadi pastikan dataset tidak terlalu besar atau gunakan sample terlebih dahulu.

#Use Cases: Classification, Regression, Clustering, dan Recommendation

#Klasifikasi dan Regresi

Klasifikasi memprediksi label kategori (churn, spam, tipe pelanggan), regresi memprediksi nilai kontinu (harga, penjualan, waktu). Keduanya mengikuti pola pipeline yang sama — hanya model dan evaluator yang berbeda.

#Clustering dengan KMeans

Untuk menemukan kelompok tersembunyi tanpa label, gunakan clustering:

KMeans clustering

from pyspark.ml.clustering import KMeans
 
kmeans = KMeans(featuresCol="features", k=4, seed=42)
model_k = kmeans.fit(data_fitur)
hasil_k = model_k.transform(data_fitur)

KMeans(featuresCol="features", k=4) membagi data menjadi empat cluster. Jumlah cluster k perlu dieksplorasi — salah satu caranya memakai teknik elbow dengan melihat cost (sum of squared distances) untuk berbagai nilai k.

#Recommendation dengan ALS

Untuk sistem rekomendasi, MLlib menyediakan ALS (Alternating Least Squares) — collaborative filtering berdasarkan interaksi user dan item:

ALS untuk rekomendasi

from pyspark.ml.recommendation import ALS
 
als = ALS(
    userCol="user_id", itemCol="produk_id", ratingCol="rating",
    coldStartStrategy="drop")
 
model_als = als.fit(interaksi)
rekomendasi = model_als.recommendForAllUsers(5)

recommendForAllUsers(5) menghasilkan lima rekomendasi teratas untuk setiap user. coldStartStrategy="drop" penting agar user atau item baru tanpa rating tidak menghasilkan null pada prediksi.

klasifikasi   → LogisticRegression, RandomForest
regresi       → LinearRegression, GBTRegressor
clustering    → KMeans, BisectingKMeans
rekomendasi   → ALS (collaborative filtering)

#Penutup

Episode 11 membekali kalian kemampuan machine learning terdistribusi: pipeline API memisahkan transformers dan estimators dengan rapi, feature engineering mengubah data mentah menjadi vektor fitur, CrossValidator melakukan tuning sistematis, dan berbagai model menangani klasifikasi, regresi, clustering, serta rekomendasi.

Inti yang harus dibawa pulang:

• Pipeline menggabungkan transformers dan estimators dalam satu alur.
• VectorAssembler dan StandardScaler adalah dasar feature engineering.
• Evaluasi selalu pada data yang tidak ikut melatih model.
• CrossValidator memilih hyperparameter secara sistematis.
• ALS menangani recommendation tanpa perlu feature manual.

Di episode 12 selanjutnya kita akan membahas security dan authentication — mengamankan cluster dengan TLS/SSL, authentication dan authorization untuk Spark jobs, mengelola kredensial data sources dengan aman, serta integrasi dengan Kerberos dan LDAP.