Machine Learning Skala Besar (Spark MLlib) & Ekosistem Big Data

Capaian Pembelajaran

Setelah minggu ini kamu bisa:

1. Menjelaskan bagaimana algoritma ML dijalankan secara data-parallel di klaster.
2. Menurunkan gradien regresi logistik dan menunjukkan bahwa gradien total = jumlah gradien per partisi.
3. Menghitung statistik (mean, std) & langkah K-Means secara terdistribusi lewat agregasi (sum, count).
4. Membangun Pipeline Spark MLlib: VectorAssembler → StandardScaler → LogisticRegression / KMeans.
5. Mengevaluasi model terdistribusi & melakukan cross-validation.
6. Memetakan komponen ekosistem big data: Hive, HBase, Kafka, Flink, data lake/lakehouse.

1. Mengapa ML Perlu Didistribusikan?

Dua jenis paralelisme:

 DATA PARALLELISM (MLlib)                    MODEL PARALLELISM (deep learning besar)
 ┌────────┐ ┌────────┐ ┌────────┐            model dipecah ke beberapa GPU,
 │part 1  │ │part 2  │ │part 3  │            data sama mengalir melewatinya
 │model w │ │model w │ │model w │
 └───┬────┘ └───┬────┘ └───┬────┘
     └── gradien/statistik ┴── dijumlah ──► driver memperbarui w ──► broadcast w baru

MLlib terutama memakai data parallelism: data dipartisi, setiap executor menghitung statistik ringkas dari partisinya, lalu driver menggabungkan.

Syarat kunci: statistik yang dihitung harus bisa dijumlahkan (additive / decomposable) — persis seperti combiner di MapReduce.

2. Regresi Logistik dengan Gradient Descent Terdistribusi

Model: p(y=1 | x) = σ(wᵀx), dengan σ(z) = 1 / (1 + e^(−z)).

Fungsi kerugian log-loss (rata-rata atas n data):

L(w) = −(1/n) Σᵢ [ yᵢ log σ(wᵀxᵢ) + (1 − yᵢ) log(1 − σ(wᵀxᵢ)) ]

Gradien:

∇L(w) = (1/n) Σᵢ (σ(wᵀxᵢ) − yᵢ) · xᵢ  =  (1/n) Xᵀ(σ(Xw) − y)

Karena berbentuk jumlah atas data, kita bisa memecahnya per partisi P₁…P_k:

Σᵢ (σ(wᵀxᵢ) − yᵢ) xᵢ  =  Σ_{i∈P₁} (…) + Σ_{i∈P₂} (…) + … + Σ_{i∈P_k} (…)
                          └─ executor 1 ─┘ └─ executor 2 ─┘     └─ executor k ─┘

Satu iterasi (gaya map-reduce):

1. driver  : broadcast w ke semua executor
2. map     : tiap partisi hitung (g_p = X_pᵀ(σ(X_p w) − y_p),  n_p)
3. reduce  : g = Σ g_p ,  n = Σ n_p           (treeAggregate di Spark)
4. driver  : w ← w − η · g / n

Hasilnya identik dengan gradient descent di satu mesin — hanya komputasinya yang dibagi. MLlib sebenarnya memakai optimizer yang lebih canggih (L-BFGS, OWL-QN) tetapi pola agregasinya sama.

Contoh hitung manual (tanpa bias): w = [0], data x = [1, 2], y = [1, 0].

• σ(0) = 0.5 untuk kedua data.
• Partisi 1 (x=1, y=1): g₁ = (0.5 − 1)·1 = −0.5
• Partisi 2 (x=2, y=0): g₂ = (0.5 − 0)·2 = 1.0
• g = −0.5 + 1.0 = 0.5, n = 2 → ∇L = 0.25
• Dengan η = 0.1: w ← 0 − 0.1·0.25 = −0.025

Tree aggregation

Jika 1000 partisi mengirim vektor gradien langsung ke driver, driver menjadi bottleneck. treeAggregate menggabungkan bertingkat:

 p1 p2 p3 p4 p5 p6 p7 p8
  \ /   \ /   \ /   \ /
  s12   s34   s56   s78        ← executor menggabungkan berpasangan
     \  /        \  /
     s1-4        s5-8
          \     /
          driver

3. Statistik Terdistribusi: StandardScaler

Mean & varian bisa dihitung dari tiga jumlah yang additive: n, Σx, Σx².

per partisi : (n_p, S1_p = Σx, S2_p = Σx²)
gabungkan   : n = Σn_p,  S1 = ΣS1_p,  S2 = ΣS2_p
mean        = S1 / n
varian      = S2 / n − mean²           (populasi, ddof=0)
std         = √varian

Contoh: partisi A = [2, 4], partisi B = [6].

• A: n=2, S1=6, S2=20. B: n=1, S1=6, S2=36.
• Total: n=3, S1=12, S2=56 → mean = 4, varian = 56/3 − 16 = 2.667, std = 1.633.

(Dalam praktik, rumus S2/n − mean² bisa kurang stabil secara numerik; Spark memakai algoritma online yang lebih stabil, misalnya varian Welford/Chan, dengan ide penggabungan yang sama.)

Catatan: StandardScaler Spark secara default tidak mengurangi mean (withMean=False) agar vektor sparse tetap sparse, dan memakai std sampel (ddof=1).

4. K-Means Terdistribusi

Satu iterasi Lloyd:

1. driver broadcast centroid  c₁…c_k
2. map   : tiap partisi, untuk tiap titik cari centroid terdekat j,
           akumulasikan  sum_j += x,  count_j += 1
3. reduce: jumlahkan sum_j dan count_j dari semua partisi
4. driver: c_j ← sum_j / count_j    (jika count_j = 0, centroid lama dipertahankan)

Inisialisasi di MLlib: k-means|| (versi paralel dari k-means++), yang memilih banyak kandidat per putaran agar tidak perlu k putaran berurutan.

5. Spark MLlib (pyspark.ml)

5a. Konsep inti (terinspirasi scikit-learn)

from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler, StandardScaler
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator

idx = StringIndexer(inputCol="kontrak", outputCol="kontrak_idx")
ohe = OneHotEncoder(inputCols=["kontrak_idx"], outputCols=["kontrak_ohe"])
asm = VectorAssembler(inputCols=["lama_langganan", "biaya_bulanan", "jumlah_keluhan",
                                 "jam_pemakaian", "kontrak_ohe"], outputCol="fitur_mentah")
sc  = StandardScaler(inputCol="fitur_mentah", outputCol="features", withMean=True)
lr  = LogisticRegression(featuresCol="features", labelCol="churn", maxIter=50, regParam=0.01)

pipe = Pipeline(stages=[idx, ohe, asm, sc, lr])
train, test = df.randomSplit([0.8, 0.2], seed=42)
model = pipe.fit(train)                  # tiap Estimator di-fit berurutan
pred  = model.transform(test)            # kolom baru: rawPrediction, probability, prediction
auc   = BinaryClassificationEvaluator(labelCol="churn").evaluate(pred)

Mengapa Pipeline penting? Statistik scaler di-fit hanya pada data latih, lalu diterapkan ke data uji → tidak ada data leakage. Seluruh pipeline bisa disimpan (model.write().save(path)) dan dipakai ulang di produksi.

5b. Tuning hyperparameter

from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
grid = ParamGridBuilder().addGrid(lr.regParam, [0.001, 0.01, 0.1]).build()
cv = CrossValidator(estimator=pipe, estimatorParamMaps=grid,
                    evaluator=BinaryClassificationEvaluator(labelCol="churn"),
                    numFolds=3, parallelism=4)
cv_model = cv.fit(train)

3 nilai × 3 fold = 9 model dilatih (+1 model final pada seluruh data latih).

5c. Algoritma yang tersedia di MLlib

5d. Kapan TIDAK perlu Spark?

Jika data muat di RAM satu mesin (< puluhan GB), scikit-learn/pandas biasanya lebih cepat & sederhana — Spark punya overhead penjadwalan dan serialisasi. Pola umum: Spark untuk ETL & fitur pada data besar, lalu sampel/agregatnya dilatih dengan scikit-learn; atau langsung MLlib bila data latih sendiri sangat besar.

6. Ekosistem Big Data

  Sumber          Ingest             Penyimpanan            Pemrosesan            Penyajian
 ┌──────┐     ┌────────────┐     ┌──────────────────┐    ┌──────────────────┐   ┌──────────┐
 │App   │────►│ Kafka      │────►│ HDFS / S3        │───►│ Spark (batch, ML)│──►│ BI/Dash- │
 │IoT   │     │ (stream)   │     │ Data lake        │    │ Flink (stream)   │   │ board    │
 │DB    │────►│ Sqoop/CDC  │     │ (Parquet, Delta, │    │ Hive (SQL)       │   │ API/ML   │
 └──────┘     └────────────┘     │  Iceberg)        │    └──────────────────┘   │ serving  │
                                 │ HBase (NoSQL)    │                           └──────────┘
                                 └──────────────────┘

Ringkasan

• MLlib = data parallelism: hitung statistik/gradien per partisi, jumlahkan, perbarui di driver.
• Gradien log-loss & statistik (n, Σx, Σx²), (sum, count) bersifat additive → bisa didistribusikan tanpa mengubah hasil.
• Pipeline (Transformer + Estimator) mencegah data leakage dan memudahkan deployment.
• Gunakan Spark ketika data benar-benar besar; jika muat di satu mesin, scikit-learn sering lebih efisien.
• Ekosistem: Kafka (ingest stream), HDFS/S3 + Parquet/Delta (simpan), Spark/Flink/Hive (olah), HBase (akses acak).

Bacaan

• Meng dkk. (2016) — MLlib: Machine Learning in Apache Spark. JMLR.
• Dokumentasi resmi: MLlib: Main Guide — ML Pipelines, Classification, Clustering.
• Leskovec, Rajaraman & Ullman — Mining of Massive Datasets, Bab 12 (Large-Scale Machine Learning).

Tugas Minggu Ini

1. Kerjakan lab.ipynb sampai selesai.
2. Kerjakan kuis (di web, atau baca quiz.json).
3. Isi exercise.py, lalu cek: python check.py 14