FECE

Machine Learning with Big Data

Instructor: Dr. Laeeq Ahmed,

Assistant Professor,
Department of Computer Science and Information Technology Jalozai,
Faculty of Electrical and Computer Engineering.
Overview
• Understand Spark MLlib core concepts: Estimators, Transformers,
Pipelines.
• Build classification, regression, and clustering workflows using
Spark MLlib.
• Learn patterns for scaling ML with Spark (data partitioning, model
parallelism, distributed hyperparameter search).
• Use MLflow to track experiments, log metrics, parameters and
models.
• Hands-on: implement a full ML pipeline (data prep → feature
engineering → model training → evaluation → MLflow tracking) in
PySpark. (Already studied in AI course)
What is MLlib?
• Spark's scalable machine learning library (built on DataFrames)
• High-level APIs for common ML tasks: classification, regression,
clustering, feature engineering
• Core concepts: Estimator, Transformer, Pipeline,
PipelineModel
• MLlib is not a drop-in replacement for scikit-learn for small data —
it targets distributed data.
 Estimators and Transformers
(concept)
 Estimator: has .fit() and produces a Transformer (example:
LogisticRegression)

 Transformer: has .transform() (example: Tokenizer, StandardScaler,

PCA)

 Pipelines: sequence of stages (mix of transformers & estimators)

raw DataFrame → transformers → [Link]() → trained model

→ transformer for predictions.
• PipelineModel: Trained pipeline that can be saved, load and reuse
(imagine how powerful is the concept)
Pipeline API (code example)
from [Link] import Pipeline
from [Link] import Tokenizer, HashingTF
from [Link] import LogisticRegression
# TRANSFORMERS: Data preprocessing steps (no training needed)
tokenizer = Tokenizer(inputCol='text', outputCol='words') # Splits text into
words
hashingTF = HashingTF(inputCol='words', outputCol='features') # Converts
words to numerical features
# ESTIMATOR: ML algorithm that needs training
lr = LogisticRegression(featuresCol='features', labelCol='label') # Classification
algorithm
# PIPELINE: Chains all steps together
pipeline = Pipeline(stages=[tokenizer, hashingTF, lr]) # Defines the
workflow
# TRAINING: Fit the entire pipeline on training data
model = [Link](train_df) # Output: PipelineModel (trained pipeline)
# PREDICTION: Use trained pipeline to make predictions on new data
preds = [Link](test_df) # Output: DataFrame with predictions
 Classification algorithms in
MLlib
 Logistic Regression, Decision Trees, RandomForest, GBT, Naive
Bayes, MultilayerPerceptron

 Multiclass support vs binary classification

 Key API patterns: .fit(), .transform(), .evaluate() (via evaluators)

 Algorithm Selection

 Parameter tuning via ParamGridBuilder and CrossValidator.

Regression algorithms in
MLlib
Built-in Loss
Algorithm Use-Cases Evaluation Metrics
Optimization
Housing prices, Sales
LinearRegression forecasting, Economic Squared Error (L2) MSE, RMSE, R², MAE
trends

Count data (web visits), Varies by family:

GeneralizedLinearRegres
Binary outcomes, Gaussian, Poisson, Deviance, AIC, RMSE
sion
Insurance claims Binomial, Gamma

Interpretable models,
DecisionTreeRegressor Medical diagnosis, Variance reduction MSE, MAE, R²
Business rules
High accuracy, Robust Average of tree
RandomForestRegressor predictions, Stock predictions (variance MSE, RMSE, R²
forecasting reduction)

Recommendation Squared Error (L2) or

GBTRegressor MSE, MAE, R²
systems Absolute Error (L1)
Clustering in MLlib
Algorithm Use Cases Key Characteristics How to Choose K

Customer segmentation,
Fast, scalable, Elbow method,
KMeans Image compression,
spherical clusters Silhouette score
Document clustering

Hierarchical data, Creates tree-like

Dendrogram analysis,
Bisecting KMeans Taxonomy creation, structure, more stable
Domain knowledge
Nested groupings than KMeans

Anomaly detection,
Overlapping clusters, Soft clustering, handles Bayesian Information
GaussianMixture
Probability-based elliptical clusters Criterion (BIC)
grouping
Spark K-means Clustering
code
from [Link] import VectorAssembler, StandardScaler
from [Link] import KMeans
from [Link] import SparkSession

spark = [Link]()

# Load some numeric data (replace with your dataset)

df = [Link]("[Link]", header=True, inferSchema=True)

# Select numeric columns for clustering

numeric_cols = ["col1", "col2", "col3"]
Spark K-means Clustering
code
# Assemble features into a vector required by Spark
assembler = VectorAssembler( inputCols=numeric_cols,
outputCol="raw_features")
df_vec = [Link](df)

# Scale features (important for Kmeans, mean 0 and SD 1 )

scaler = StandardScaler(inputCol="raw_features",
outputCol="features",
withMean=True,
withStd=True
)
df_scaled = [Link](df_vec).transform(df_vec)
Spark K-means Clustering
code
# Train KMeans
kmeans = KMeans(k=4, seed=42, featuresCol="features")
model = [Link](df_scaled)

preds = [Link](df_scaled) #Predictions

# Evaluate with Silhouette Score to find no. of clusters

from [Link] import ClusteringEvaluator
evaluator = ClusteringEvaluator(featuresCol="features",
predictionCol="prediction"
)
silhouette = [Link](preds)
Spark K-means Clustering
code
print("Silhouette Score:", silhouette)
print("Cluster Centers:")
for c in [Link]():
print(c)
 Feature engineering &
feature transformers
 Common transformers: StringIndexer, OneHotEncoder, VectorAssembler,
StandardScaler, PCA, HashingTF, CountVectorizer.

 VectorAssembler example: columns to vector [25.0, 50000.0, 85.0]

 Handling categorical variables, text, and vectors.

• Feature engineering is often the most important work in ML pipelines.

Model Selection & Hyperparameter
Tuning in Spark MLlib
Component Purpose Key Parameters

Defines hyperparameter search

ParamGridBuilder .addGrid(), .build()
space

HyperPTuning, Robust k-fold numFolds=3, seed, paralle

CrossValidator
cross-validation lism
TrainValidationSpl HyperPTuning, Faster trainRatio=0.75, seed, par
it train/validation split allelism

BinaryClassificationEvalua
Evaluators Measures model performance tor, RegressionEvaluator,
ClusteringEvaluator
Example: end-to-end
classification pipeline (code)
from [Link] import Pipeline
from [Link] import StringIndexer, VectorAssembler,
StandardScaler
from [Link] import RandomForestClassifier
# preprocessing
si = StringIndexer(inputCol='cat', outputCol='cat_idx')
va = VectorAssembler(inputCols=['num1','num2','cat_idx'],
outputCol='raw_features')
sc = StandardScaler(inputCol='raw_features', outputCol='features')
rf = RandomForestClassifier(labelCol='label', featuresCol='features')
pipe = Pipeline(stages=[si, va, sc, rf])
model = [Link](train_df)

• Take care of Reproducibility and Saving Pipeline, seed is missing in

RandomForest which can help with Reproducibility.
Grid search with CrossValidator
from [Link] import ParamGridBuilder, CrossValidator

from [Link] import BinaryClassificationEvaluator

# Create an evaluator to measure model performance during tuning

# This will use AUC (Area Under ROC Curve) by default for binary classification

evaluator = BinaryClassificationEvaluator(labelCol="label")

# Define the hyperparameter search space using ParamGridBuilder

# This creates a grid of all possible parameter combinations to test

paramGrid = (ParamGridBuilder()

.addGrid([Link], [20, 50]) # Test with 20 and 50 trees in the forest

.addGrid([Link], [5, 10]) # Test with max depth of 5 and 10 levels

.build() # Finalizes the grid: 2 × 2 = 4 combinations

)
Grid search with CrossValidator
# Set up CrossValidator for k-fold cross-validation with hyperparameter tuning
cv = CrossValidator(
estimator=pipeline, # The ML pipeline to tune (includes data prep + model)
estimatorParamMaps=paramGrid, # The 4 parameter combinations to test
evaluator=evaluator, # How to score each model (using AUC)
numFolds=3, # Use 3-fold cross-validation for robustness
parallelism=4 # Run 4 parameter combinations in parallel for speed
)
# Execute the hyperparameter tuning process
# This will train 4 param combinations × 3 folds = 12 total models
cvModel = [Link](train_df)
# Extract the best performing model from all combinations tested
bestModel = [Link] # This is the model with optimal numTrees and maxDepth
# The bestModel can now be used for predictions on new data
predictions = [Link](test_df)
Break / Quick Q&A
• Any Questions

• Implementation Issues??
Advanced Spark ML: Scaling and Manageme
Tips and Tricks

1. Data Handling at Scale

2. Pipeline execution and Management
Scaling ML with Spark: Why It
Matters?
Key Idea: Large datasets break single-machine ML; Spark distributes both data and computation.
Challenges in large-scale ML
• Data too large for memory
• Long training times on a single node
• Hyperparameter tuning becomes infeasible (multiple combinations)
• Preprocessing (transformations) must be applied consistently at scale

How Spark solves this

• Distributed DataFrames (RDD-backed execution)
• MLlib algorithms optimized for cluster execution
• Pipelines ensure consistent transformations everywhere
• Built-in parallelism for cross-validation and grid search
Takeaway
Scaling ML isn’t only “running faster”; it’s ensuring consistent, repeatable, cluster-wide workflows.
Data Partitioning & Its Impact
on ML
Partitioning principles
• Spark ML works on partitions — number & size affect performance
• Too few partitions → underutilized cluster (partitions>>cores)
• Too many partitions → overhead, shuffles explode (TCP connections,
metadata, etc.)

Good practices
• Repartition by key for better distribution
• Cache data before iterative ML algorithms
• Reduce skew (unbalanced partitioning) before training (salting, bucketing)
Data Partitioning & Its Impact
on ML
# Inspect partition count
print(train_df.[Link]())

# Repartition by a feature (common in joins + ML prep)

train_df = train_df.repartition(200, "label")

# Cache for iterative algorithms

train_df = train_df.cache()
Handling Skew (Hot Keys) in
ML Pipelines
Skew kills performance during feature prep, joins, and pipeline
execution.
Signs of skew
• One executor doing most work
• Very slow stages with “task 0” lagging in the UI
• Huge shuffle files
Fixing key skew (salting)
from [Link] import functions as F
salt_size = 10 # number of buckets
small_salted = small_df.withColumn("salt",
[Link]([Link]([[Link](i) for i in range(salt_size)])))
big_salted = big_df.withColumn("salt",
([Link]()*salt_size).cast("int"))
joined = big_salted.join(small_salted, ["id", "salt"])

BEFORE SALTING: Big Table: (userA, data1), (userA, data2), (userA, data3) ← All in same partition
Small Table: (userA, metadata) ← One copy

AFTER SALTING:
Big Table: (userA, salt0, data1), (userA, salt1, data2), (userA, salt2, data3)
Small Table: (userA, salt0, metadata), (userA, salt1, metadata), (userA, salt2, metadata)

Now joins happen per-salt bucket, distributing userA's data!

Distributed Model Training in
Spark MLlib
What scales naturally (Data Parallelism vs Model Parallelism)
• Linear regression
• Logistic regression
• Decision trees / Random forests
• Gradient boosted trees (tree-parallel, not GPU)
• KMeans & clustering algorithms

Spark follows Distributed training model

• Data split across workers
• Each worker computes partial statistics
• Aggregation on driver or tree-coordinator
• Model parameters update globally

Tip Algorithms like ANN, CNN, transformers → not Spark MLlib territory. (Use Pytorch or tensonflow)
Distributed Hyperparameter
Search
Spark performs CV and grid search in parallel across executors.
• Pipeline-friendly (pipeline can be given as estimator)
• Wrap everything in a Pipeline
• We can provide hyperparameters of final estimator in the gridform
• Use CrossValidator/TrainValidationSplit

from [Link] import ParamGridBuilder, CrossValidator

from [Link] import BinaryClassificationEvaluator

paramGrid = (ParamGridBuilder()
.addGrid([Link], [5, 10])
.addGrid([Link], [50, 100])
.build())
Distributed Hyperparameter
Search
cv = CrossValidator(
estimator=pipeline,
estimatorParamMaps=paramGrid,
evaluator=BinaryClassificationEvaluator(), numFolds=3,
parallelism=6
)
cvModel = [Link](train_df)

Key scaling tricks

• Increase parallelism
• Avoid huge grids
• Cache training data before CV
MLflow: Why You Need
Experiment Tracking?
Motivations
• Hard to track which model used which features
• Hard to reproduce hyperparameters
• Metrics get lost
• Saving models manually becomes messy

MLflow solves
• Versioned parameters and Versioned metrics
• Versioned artifacts (plots, datasets, models)
• Lineage: pipeline → model → run → experiment

Integrates directly with Spark

• MLflow can log Spark pipelines
• Can load Spark models for inference
MLflow Essentials (Runs,
Experiments, Tracking)
Concepts
• Experiment: A project (e.g., "Credit Risk Models")
• Run: A single training execution
• Artifacts: Plots, models, feature importances
• Metrics: AUC, accuracy, loss
• Parameters: Hyperparameters, pipeline settings

Basic workflow
[Link].start_run() [Link] params
[Link] metrics [Link] model
[Link] ends → everything saved

MLflow doesn’t store your DataFrame

Only metadata, models, and logs.
MLflow + Spark (Short Code
Example)
import mlflow
import [Link]
with mlflow.start_run():
model = [Link](train_df)
mlflow.log_param("maxDepth", 10)
mlflow.log_metric("train_auc",
[Link]([Link](train_df)))

# Save entire Spark PipelineModel

[Link].log_model(model, "model")

print("Run saved:", mlflow.active_run())

Why this matters

•The exact trained pipeline is preserved
•Can be deployed / loaded without retraining
 Reproducibility: Controlling
Randomness + Versioning
Steps to guarantee reproducibility
• Set random seeds for MLlib models
• Save the PipelineModel
• Track everything via MLflow
• Version your code + data transformations
Example
rf = RandomForestClassifier(seed=42)
Pipeline saving
[Link]().overwrite().save("models/final_pipeline")
Loading
from [Link] import PipelineModel
model = [Link]("models/final_pipeline")
Hyperparameter Search +
MLflow
import mlflow, [Link]
from [Link] import ParamGridBuilder, CrossValidator

paramGrid = (ParamGridBuilder()
.addGrid([Link], [50, 100])
.addGrid([Link], [5, 10])
.build())

cv = CrossValidator(estimator=pipeline, estimatorParamMaps=paramGrid,
evaluator=BinaryClassificationEvaluator(), numFolds=3)

with mlflow.start_run():
cvModel = [Link](train_df)
best = [Link]
[Link].log_model(best, "best_model")
mlflow.log_param("grid_size", len(paramGrid))
Deploying the Saved Spark
Model
Load the MLflow-logged model
loaded = [Link].load_model("runs:/<run_id>/best_model")
preds = [Link](new_data)

Why this matters

• No need to rerun training
• Consistent preprocessing (indexing, scaling included)
• Reproducible scoring pipeline
Hands-On Pipeline Summary
Students will build
1. Load + clean data
2. Handle missing values
3. Feature transformations
4. VectorAssembler + StandardScaler
5. Train a model (RF or LR)
6. Use CrossValidator
7. Track run with MLflow
8. Save + load pipeline

Goal
A full end-to-end, reproducible, scalable pipeline.
 Summary
Today we have covered:
• Data partitioning, skew handling, distributed training
• Hyperparameter tuning with Spark CV
• Experiment tracking with MLflow
• Saving & loading full Spark ML Pipelines
• Implementing an end-to-end, production-ready ML workflow

Outcome
Students should be able to build scalable, tracked, reproducible, Spark-native ML
systems.
Thanks
o Email: laeeq@[Link]
o Any Questions