PySpark untuk Machine Learning

Daftar Isi

Pendahuluan

Apache Spark adalah standar de facto untuk pemrosesan data terdistribusi berskala besar. Library machine learning-nya, MLlib, menyediakan implementasi algoritma ML umum yang dapat diskalakan dan berjalan mulus pada klaster yang terdiri dari ratusan mesin.

PySpark, yaitu API Python untuk Spark, memungkinkan data scientist menggunakan sintaksis Python yang familiar sembari memanfaatkan kekuatan penuh mesin komputasi terdistribusi Spark. Tutorial ini mencakup alur kerja ML PySpark secara lengkap: dari membuat SparkSession dan memanipulasi DataFrame, melalui pembangunan model klasifikasi, regresi, dan klasterisasi, hingga deployment model dengan Spark Structured Streaming.

Baik Anda memproses data berukuran gigabyte maupun terabyte, pola-pola dalam tutorial ini berlaku sama.

Prasyarat

• Python 3.9+
• Java 11 atau 17 (diperlukan oleh Spark)
• Pemahaman dasar tentang konsep machine learning

Instalasi PySpark:

pip install pyspark numpy pandas

Untuk deployment di klaster, pastikan Spark terinstal di semua node. Untuk pengembangan lokal, PySpark menyertakan instance Spark standalone bawaan.

Dasar Spark dan SparkSession

Setiap aplikasi PySpark dimulai dengan SparkSession, yang merupakan titik masuk terpadu untuk semua fungsionalitas Spark.

Membuat SparkSession

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("Tutorial ML PySpark") \
    .master("local[]") \

    .config("spark.driver.memory", "4g") \
    .config("spark.sql.shuffle.partitions", "8") \
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") \
    .getOrCreate()

Verifikasi sesi
print(f"Versi Spark: {spark.version}")
print(f"Nama aplikasi: {spark.sparkContext.appName}")
print(f"Master: {spark.sparkContext.master}")
print(f"Paralelisme default: {spark.sparkContext.defaultParallelism}")

from pyspark.sql.types import StructType, StructField, FloatType, IntegerType, StringType
import numpy as np

Membuat dari data Python
data = [
    (1, "Alice", 28, 55000.0, "Engineering"),
    (2, "Bob", 35, 72000.0, "Marketing"),
    (3, "Carol", 42, 88000.0, "Engineering"),
    (4, "Dave", 31, 61000.0, "Sales"),
    (5, "Eve", 26, 48000.0, "Marketing"),
]

skema = StructType([
    StructField("id", IntegerType(), False),
    StructField("nama", StringType(), False),
    StructField("umur", IntegerType(), False),
    StructField("gaji", FloatType(), False),
    StructField("departemen", StringType(), False),
])

df = spark.createDataFrame(data, schema=skema)
df.show()
df.printSchema()

Membaca dari CSV
dfcsv = spark.read.csv("data.csv", header=True, inferSchema=True)


Membaca dari Parquet (format kolom yang dioptimalkan)
dfparquet = spark.read.parquet("data.parquet")

Membaca dari JSON
dfjson = spark.read.json("data.json")

from pyspark.sql import functions as F

Select dan filter
df.select("nama", "umur", "gaji").filter(F.col("umur") > 30).show()

Menambahkan kolom komputasi
df = df.withColumn("gajibulanan", F.col("gaji") / 12)
df = df.withColumn("kelompokumur", F.when(F.col("umur") < 30, "Muda")

                                      .when(F.col("umur") < 40, "Menengah")
                                      .otherwise("Senior"))

Agregasi
df.groupBy("departemen").agg(
    F.avg("gaji").alias("ratagaji"),
    F.count("").alias("jumlah"),
    F.max("umur").alias("umurmaks"),

).show()

Menangani nilai yang hilang
dfbersih = df.na.fill({"gaji": 0.0, "departemen": "Tidak Diketahui"})
dfbersih = df.na.drop(subset=["umur", "gaji"])


Pengurutan
df.orderBy(F.col("gaji").desc()).show()

Statistik
df.describe("umur", "gaji").show()
print(f"Korelasi (umur, gaji): {df.stat.corr('umur', 'gaji'):.4f}")

from pyspark.ml.feature import StringIndexer, OneHotEncoder

Membuat dataset sintetis yang lebih besar
np.random.seed(42)
departemenlist = ["Engineering", "Marketing", "Sales", "HR", "Finance"]
baris = [(int(i), str(np.random.choice(departemenlist)),

          int(np.random.randint(22, 60)),
          float(np.random.uniform(30000, 120000)),
          int(np.random.choice([0, 1])))
         for i in range(10000)]

skemaml = StructType([
    StructField("id", IntegerType()), StructField("departemen", StringType()),
    StructField("umur", IntegerType()), StructField("gaji", FloatType()),
    StructField("label", IntegerType()),
])
dfml = spark.createDataFrame(baris, skemaml)

String indexing
deptindexer = StringIndexer(inputCol="departemen", outputCol="deptindex")
dfml = deptindexer.fit(dfml).transform(dfml)

One-hot encoding
deptencoder = OneHotEncoder(inputCol="deptindex", outputCol="deptonehot")

dfml = deptencoder.fit(dfml).transform(dfml)


dfml.select("departemen", "deptindex", "deptonehot").show(5, truncate=False)

Penskalaan dan Normalisasi Fitur Numerik

from pyspark.ml.feature import VectorAssembler, StandardScaler, MinMaxScaler

Menggabungkan fitur numerik menjadi satu vektor
assembler = VectorAssembler(
    inputCols=["umur", "gaji"],
    outputCol="fiturnumerik"

)
dfml = assembler.transform(dfml)


Penskalaan standar (rata-rata nol, variansi satu)
stdscaler = StandardScaler(
    inputCol="fiturnumerik",

    outputCol="fiturterskalakan",
    withMean=True,
    withStd=True,
)
modelscaler = stdscaler.fit(dfml)

dfml = modelscaler.transform(dfml)

Penskalaan MinMax
minmaxscaler = MinMaxScaler(inputCol="fiturnumerik", outputCol="fiturminmax")

dfml = minmaxscaler.fit(dfml).transform(dfml)

from pyspark.ml.feature import Bucketizer, PCA, PolynomialExpansion

Bucketizing fitur kontinu
bucketizer = Bucketizer(
    splits=[0, 25, 35, 45, 100],
    inputCol="umur",
    outputCol="bucketumur"
)
dfml = bucketizer.transform(dfml)

PCA untuk reduksi dimensi
pca = PCA(k=2, inputCol="fiturterskalakan", outputCol="fiturpca")
modelpca = pca.fit(dfml)
dfml = modelpca.transform(dfml)

print(f"Variansi terjelaskan PCA: {modelpca.explainedVariance}")

Ekspansi polinomial
poly = PolynomialExpansion(degree=2, inputCol="fiturnumerik", outputCol="fiturpoly")
dfml = poly.transform(dfml)

Klasifikasi dengan Spark MLlib

Regresi Logistik

from pyspark.ml.feature import VectorAssembler
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator, MulticlassClassificationEvaluator

Menyiapkan fitur
kolomfitur = ["umur", "gaji", "deptindex"]
assembler = VectorAssembler(inputCols=kolomfitur, outputCol="features")

dfsiap = assembler.transform(dfml)


Membagi data
dflatih, dfuji = dfsiap.randomSplit([0.8, 0.2], seed=42)

Melatih regresi logistik
lr = LogisticRegression(
    featuresCol="features",
    labelCol="label",
    maxIter=100,
    regParam=0.01,
    elasticNetParam=0.8,
)
modellr = lr.fit(dflatih)

Prediksi
prediksi = modellr.transform(dfuji)
prediksi.select("label", "prediction", "probability").show(10, truncate=False)

Evaluasi
evalbiner = BinaryClassificationEvaluator(labelCol="label")

auc = evalbiner.evaluate(prediksi, {evalbiner.metricName: "areaUnderROC"})


evalmulti = MulticlassClassificationEvaluator(labelCol="label")
akurasi = evalmulti.evaluate(prediksi, {evalmulti.metricName: "accuracy"})
f1 = evalmulti.evaluate(prediksi, {evalmulti.metricName: "f1"})

print(f"AUC-ROC:  {auc:.4f}")
print(f"Akurasi:  {akurasi:.4f}")
print(f"Skor F1:  {f1:.4f}")

Random Forest Classifier

from pyspark.ml.classification import RandomForestClassifier

rf = RandomForestClassifier(
    featuresCol="features",
    labelCol="label",
    numTrees=100,
    maxDepth=10,
    featureSubsetStrategy="sqrt",
    seed=42,
)
modelrf = rf.fit(dflatih)

prediksirf = modelrf.transform(dfuji)

akurasirf = evalmulti.evaluate(prediksirf, {evalmulti.metricName: "accuracy"})

print(f"Akurasi Random Forest: {akurasirf:.4f}")

Pentingnya fitur
pentingnya = modelrf.featureImportances.toArray()

for kolom, nilai in zip(kolomfitur, pentingnya):
    print(f"  {kolom}: {nilai:.4f}")

Gradient-Boosted Trees

from pyspark.ml.classification import GBTClassifier

gbt = GBTClassifier(
    featuresCol="features",
    labelCol="label",
    maxIter=50,
    maxDepth=5,
    stepSize=0.1,
    seed=42,
)
modelgbt = gbt.fit(dflatih)

prediksigbt = modelgbt.transform(dfuji)

akurasigbt = evalmulti.evaluate(prediksigbt, {evalmulti.metricName: "accuracy"})

print(f"Akurasi GBT: {akurasigbt:.4f}")

Regresi dengan Spark MLlib

Regresi Linear

from pyspark.ml.regression import LinearRegression
from pyspark.ml.evaluation import RegressionEvaluator

Menyiapkan data regresi (memprediksi gaji dari umur dan departemen)
assemblerreg = VectorAssembler(inputCols=["umur", "deptindex"], outputCol="features")
dfreg = assemblerreg.transform(dfml)


latihreg, ujireg = dfreg.randomSplit([0.8, 0.2], seed=42)

reglinear = LinearRegression(

    featuresCol="features",
    labelCol="gaji",
    maxIter=100,
    regParam=0.01,
    elasticNetParam=0.5,
)
modelreg = reglinear.fit(latihreg)

print(f"Koefisien: {modelreg.coefficients}")

print(f"Intercept: {modelreg.intercept:.2f}")

prediksireg = modelreg.transform(ujireg)


evalreg = RegressionEvaluator(labelCol="gaji")
rmse = evalreg.evaluate(prediksireg, {evalreg.metricName: "rmse"})

mae = evalreg.evaluate(prediksireg, {evalreg.metricName: "mae"})
r2 = evalreg.evaluate(prediksireg, {evalreg.metricName: "r2"})


print(f"RMSE: {rmse:.2f}")
print(f"MAE:  {mae:.2f}")
print(f"R2:   {r2:.4f}")

from pyspark.ml.regression import RandomForestRegressor

rfreg = RandomForestRegressor(
    featuresCol="features",
    labelCol="gaji",
    numTrees=100,
    maxDepth=10,
    seed=42,
)
modelrfreg = rfreg.fit(latihreg)
predrfreg = modelrfreg.transform(ujireg)


rmserf = evalreg.evaluate(predrfreg, {evalreg.metricName: "rmse"})
print(f"RMSE Regresi RF: {rmserf:.2f}")

from pyspark.ml.clustering import KMeans
from pyspark.ml.evaluation import ClusteringEvaluator

assemblerklaster = VectorAssembler(inputCols=["umur", "gaji"], outputCol="features")
dfklaster = assemblerklaster.transform(dfml)


Penskalaan standar fitur untuk klasterisasi
from pyspark.ml.feature import StandardScaler
scalerklaster = StandardScaler(inputCol="features", outputCol="fiturterskalakan", withMean=True, withStd=True)

dfklaster = scalerklaster.fit(dfklaster).transform(dfklaster)


Mencari k optimal menggunakan skor silhouette
skorsilhouette = []
for k in range(2, 8):
    kmeans = KMeans(featuresCol="fiturterskalakan", k=k, seed=42, maxIter=50)

    modelkm = kmeans.fit(dfklaster)

    prediksikm = modelkm.transform(dfklaster)
    evaluator = ClusteringEvaluator(featuresCol="fiturterskalakan")

    skor = evaluator.evaluate(prediksikm)
    skorsilhouette.append((k, skor))

    print(f"k={k}: Skor Silhouette = {skor:.4f}")

Melatih dengan k optimal
koptimal = max(skorsilhouette, key=lambda x: x[1])[0]

print(f"\nk optimal: {koptimal}")

kmeansfinal = KMeans(featuresCol="fiturterskalakan", k=koptimal, seed=42)

modelkmfinal = kmeansfinal.fit(dfklaster)


Pusat klaster
for i, pusat in enumerate(modelkmfinal.clusterCenters()):

    print(f"Klaster {i}: {pusat}")

from pyspark.ml.clustering import GaussianMixture

gmm = GaussianMixture(featuresCol="fiturterskalakan", k=3, seed=42, maxIter=100)
modelgmm = gmm.fit(dfklaster)

prediksigmm = modelgmm.transform(dfklaster)

prediksigmm.select("umur", "gaji", "prediction", "probability").show(10, truncate=False)

Pipeline API

Pipeline Spark ML merangkai beberapa transformer dan estimator menjadi satu alur kerja yang dapat direproduksi.

Membangun Pipeline Lengkap

from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler, StandardScaler
from pyspark.ml.classification import RandomForestClassifier

Mendefinisikan tahapan pipeline
deptindexer = StringIndexer(inputCol="departemen", outputCol="deptindex", handleInvalid="keep")
deptencoder = OneHotEncoder(inputCol="deptindex", outputCol="deptencoded")

assembler = VectorAssembler(inputCols=["umur", "gaji", "deptencoded"], outputCol="fiturmentah")

scaler = StandardScaler(inputCol="fiturmentah", outputCol="features", withMean=False, withStd=True)
classifier = RandomForestClassifier(
    featuresCol="features", labelCol="label", numTrees=100, maxDepth=10, seed=42
)

Membuat dan melatih pipeline
pipeline = Pipeline(stages=[deptindexer, deptencoder, assembler, scaler, classifier])

latihmentah, ujimentah = dfml.randomSplit([0.8, 0.2], seed=42)

modelpipeline = pipeline.fit(latihmentah)


Prediksi menggunakan pipeline lengkap
prediksipipeline = modelpipeline.transform(ujimentah)
prediksipipeline.select("departemen", "umur", "gaji", "label", "prediction").show(10)


akurasi = MulticlassClassificationEvaluator(labelCol="label").evaluate(
    prediksipipeline, {MulticlassClassificationEvaluator.metricName: "accuracy"}
)
print(f"Akurasi Pipeline: {akurasi:.4f}")

Validasi Silang dan Penyetelan Hyperparameter

Grid Search dengan Validasi Silang

from pyspark.ml.tuning import CrossValidator, ParamGridBuilder

Membangun grid parameter
gridparam = ParamGridBuilder() \

    .addGrid(classifier.numTrees, [50, 100, 200]) \
    .addGrid(classifier.maxDepth, [5, 10, 15]) \
    .addGrid(classifier.featureSubsetStrategy, ["sqrt", "log2"]) \
    .build()

print(f"Total kombinasi parameter: {len(gridparam)}")

Membuat cross-validator
evaluator = BinaryClassificationEvaluator(labelCol="label", metricName="areaUnderROC")

cv = CrossValidator(
    estimator=pipeline,
    estimatorParamMaps=gridparam,

    evaluator=evaluator,
    numFolds=5,
    parallelism=4,       # evaluasi 4 model secara paralel
    seed=42,
)

Melatih cross-validator
modelcv = cv.fit(latihmentah)


Hasil model terbaik
modelterbaik = modelcv.bestModel

rfterbaik = modelterbaik.stages[-1]  # tahap terakhir adalah classifier

print(f"numTrees terbaik: {rfterbaik.getNumTrees}")
print(f"maxDepth terbaik: {rfterbaik.getOrDefault('maxDepth')}")


prediksicv = modelcv.transform(ujimentah)
auccv = evaluator.evaluate(prediksicv)
print(f"AUC Validasi Silang: {auccv:.4f}")


Semua metrik fold
print("\nMetrik rata-rata per kombinasi parameter:")
for i, metrik in enumerate(modelcv.avgMetrics):
    print(f"  Kombinasi {i}: AUC = {metrik:.4f}")

Train-Validation Split (Alternatif Lebih Cepat)

from pyspark.ml.tuning import TrainValidationSplit

tvs = TrainValidationSplit(
    estimator=pipeline,
    estimatorParamMaps=gridparam,

    evaluator=evaluator,
    trainRatio=0.8,
    parallelism=4,
    seed=42,
)

modeltvs = tvs.fit(latihmentah)

prediksitvs = modeltvs.transform(ujimentah)
auctvs = evaluator.evaluate(prediksitvs)
print(f"AUC TVS: {auctvs:.4f}")

# Menyimpan seluruh model pipeline
modelpipeline.write().overwrite().save("models/rfpipeline")


Menyimpan model terbaik dari validasi silang
modelcv.bestModel.write().overwrite().save("models/modelcvterbaik")

Memuat model
from pyspark.ml import PipelineModel

pipelinedimuat = PipelineModel.load("models/rfpipeline")
prediksidimuat = pipelinedimuat.transform(ujimentah)

akurasidimuat = MulticlassClassificationEvaluator(labelCol="label").evaluate(
    prediksidimuat, {MulticlassClassificationEvaluator.metricName: "accuracy"}

)
print(f"Akurasi model yang dimuat: {akurasidimuat:.4f}")

Mengekspor untuk Lingkungan Non-Spark

# Mengonversi model Spark ML ke PMML atau ONNX untuk deployment di luar Spark
Opsi 1: Mengekstrak parameter model secara manual
modelrffinal = modelpipeline.stages[-1]

print(f"Jumlah pohon: {modelrffinal.getNumTrees}")

print(f"Pentingnya fitur: {modelrffinal.featureImportances}")


Opsi 2: Menggunakan MLeap untuk serialisasi portabel
pip install mleap
from mleap.pyspark.sparksupport import SimpleSparkSerializer
modelpipeline.serializeToBundle("jar:file:/path/to/model.zip", ujimentah)

Deployment dengan Spark Structured Streaming

Spark Structured Streaming memungkinkan Anda menerapkan model ML terlatih pada aliran data waktu nyata.

Pipeline Inferensi Streaming

from pyspark.sql.types import StructType, StructField, StringType, IntegerType, FloatType

Mendefinisikan skema untuk data stream masuk
skemastream = StructType([

    StructField("id", IntegerType()),
    StructField("departemen", StringType()),
    StructField("umur", IntegerType()),
    StructField("gaji", FloatType()),
])

Membaca dari sumber streaming (misal: file yang masuk ke direktori)
dfstreaming = spark.readStream \
    .schema(skemastream) \

    .option("maxFilesPerTrigger", 1) \
    .json("data/inputstream/")

Menerapkan model pipeline terlatih pada data streaming
prediksistreaming = pipelinedimuat.transform(dfstreaming)


Menulis prediksi ke output
query = prediksistreaming \
    .select("id", "departemen", "umur", "gaji", "prediction") \
    .writeStream \
    .outputMode("append") \
    .format("json") \
    .option("path", "data/outputstream/") \

    .option("checkpointLocation", "data/checkpoint/") \
    .trigger(processingTime="10 seconds") \
    .start()

Memantau stream
print(f"Stream aktif: {query.isActive}")
print(f"Status stream: {query.status}")

query.awaitTermination()  # uncomment untuk berjalan tanpa batas
query.stop()  # menghentikan stream

# Membaca dari Kafka
dfkafka = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "ml-input") \
    .load()

Memparse nilai Kafka (string JSON) menjadi data terstruktur
from pyspark.sql.functions import fromjson, col


dfterparsing = dfkafka.select(

    fromjson(col("value").cast("string"), skemastream).alias("data")

).select("data.")


Menerapkan model ML
prediksikafka = pipelinedimuat.transform(dfterparsing)

Menulis prediksi kembali ke Kafka
query = prediksikafka.select(

    col("id").cast("string").alias("key"),
    col("prediction").cast("string").alias("value"),
).writeStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("topic", "ml-output") \
    .option("checkpointLocation", "data/kafkacheckpoint/") \
    .start()

dflatih.cache()
print(f"Partisi ter-cache: {dflatih.rdd.getNumPartitions()}")

from pyspark.sql.functions import broadcast
hasil = dfbesar.join(broadcast(dfkecil), on="kunci")

Kesimpulan

Tutorial ini telah membahas alur kerja machine learning PySpark secara lengkap:

• SparkSession dan DataFrame menyediakan fondasi untuk pemrosesan data terdistribusi dengan API mirip SQL.
• Rekayasa fitur dalam skala besar ditangani oleh transformer MLlib termasuk StringIndexer, OneHotEncoder, VectorAssembler, dan StandardScaler.
• Algoritma klasifikasi, regresi, dan klasterisasi di MLlib dapat diskalakan secara horizontal di seluruh klaster sambil mempertahankan antarmuka yang familiar mirip scikit-learn.
• Pipeline merangkai semua langkah preprocessing dan pemodelan menjadi satu objek yang dapat direproduksi dan diserialisasi.
• Validasi silang dengan ParamGridBuilder memungkinkan penyetelan hyperparameter yang sistematis dengan evaluasi paralel.
• Persistensi model memungkinkan penyimpanan dan pemuatan model pipeline lengkap untuk deployment produksi.
• Structured Streaming memungkinkan inferensi waktu nyata dengan menerapkan model terlatih pada aliran data langsung dari file atau Kafka.

PySpark adalah alat yang tepat ketika data Anda melampaui kapasitas satu mesin atau ketika Anda memerlukan manfaat operasional dari ekosistem Spark (toleransi kesalahan, streaming, dan integrasi dengan Hadoop, Hive, dan Kafka). Untuk dataset yang lebih kecil, scikit-learn tetap lebih sederhana dan lebih cepat. Kuncinya adalah memilih alat yang tepat untuk skala masalah Anda.