Spark MLlib Architecture, Feature Engineering Pipelines, and Distributed Model Training

15.1 Spark MLlib Overview, Scalability, and Architectural Foundations

Machine learning over massive datasets needs a compute fabric that grows by adding machines, not by buying a faster single box. Traditional single-node libraries (scikit-learn, local R, a lone GPU workstation) run the training loop on one machine. Physical core count on that machine is hard-capped — typically between 64 and 128 cores even on high-end hardware. Apache Spark MLlib (Spark’s scalable machine learning library) sits on distributed clusters of commodity workers and grows horizontally: add worker nodes, gain aggregate cores, memory, and disk. Motivation for Spark MLlib is exactly this scalability trade-off: accept cluster coordination cost to train on data that no single node can hold.

Companion texts place MLlib in the Spark stack alongside Spark SQL, Streaming, and GraphX: it ships classification, regression, clustering, collaborative filtering, dimensionality reduction, and optimization primitives for recommendation, clustering, and classification workloads. Practitioner references such as Spark: The Definitive Guide by Matei Zaharia and Bill Chambers describe the same DataFrame-centric ML pipeline model used in modern pyspark.ml.

15.1.1 Horizontal Scaling versus Single-Node Optimization

In production ML engineering the standing choice is: spend effort optimizing a model on one machine, or spend effort scaling the cluster so larger datasets fit. Prefer scaling. Enterprise data volumes grow every year; model quality and rare-pattern detection increasingly depend on how much training data you can ingest, not on another 5% of local FLOPS.

Visual intuition. Imagine a bar chart: horizontal axis = number of worker nodes (\(N\)), vertical axis = aggregate cores (\(Nc\)). With fixed \(c=4\), the plot is a straight line through the origin with slope 4. Landmarks: \(N=100 \rightarrow 400\) cores; \(N=150 \rightarrow 600\) cores. Takeaway: capacity grows linearly with nodes under a fixed per-node core count.

Production systems prefer scaling distributed hardware over local algorithm micro-optimization because annual enterprise data growth demands multi-terabyte ingestion.

15.1.2 Evolution of Spark MLlib API Packages

Apache Spark exposes ML through two package namespaces:

Spark’s design also stresses pipeline composition: data leaves sources, passes through ML stages, and feeds applications. MLlib APIs interoperate with Spark SQL and, in Python, with NumPy-style numeric arrays — useful when assembling feature vectors before estimators run.

15.1.3 Scalability Limits and Feature Volume Capacities

Spark MLlib algorithms target high-dimensional feature spaces and row counts that exceed single-node RAM:

• Decision Trees: up to about 1,000 continuous or categorical features, with information-gain / entropy splits computed across partitions.
• Random Forest: up to about 10,000 features and classification with up to about 100,000 (100k) distinct output classes.
• Logistic Regression: sparse or dense feature spaces up to about 10,000,000 (10M) features.
• Training instances (rows): effectively unbounded for supervised and unsupervised algorithms — limited by aggregate cluster storage and memory, not by a fixed library constant.

Real-world & domain. Clickstream and ad-tech logistic models routinely use millions of sparse hashed or one-hot features across a Spark cluster; genomics and retail assortment models use random forests over thousands of engineered attributes. MLlib exists so these workloads stay inside the same Spark SQL / DataFrame fabric used for ETL.

Architecture and scale motivate why Spark ML exists. The next section defines the four-stage ML lifecycle and the Transformer / Estimator / Evaluator / Pipeline abstractions that implement it.

15.2 Spark ML Machine Learning Pipeline Architecture

A complete machine learning workflow spans raw data loading through scoring and validation. Spark MLlib formalizes this lifecycle as a modular pipeline: each stage is a reusable Transformer or Estimator, and the Pipeline class chains them into one executable workflow. Companion material describes an ML pipeline as data taken from sources, passed through ML programs, with outputs feeding applications (decision trees, clustering, mining, and related tasks).

15.2.1 The Four Stages of the Machine Learning Lifecycle

These stages align with classical ETL thinking (extract → transform → load/model): Spark folds the “transform” of business data and the “transform” of ML features into the same DataFrame API.

15.2.2 Core Pipeline Abstractions: Transformers, Estimators, and Evaluators

A Pipeline is represented by the Pipeline class: stages execute in order; fitting a pipeline fits every Estimator stage and produces a PipelineModel whose stages are all Transformers.

15.2.3 Vector Representations: Dense Vectors versus Sparse Vectors

Spark MLlib algorithms require all input features in one vector column of double-precision floats (Vector of Double). Labels must also be Double.

Spark provides two implementations of the Vector interface in pyspark.ml.linalg:

Comparison — when to pick which.

When to pick which: use dense for low-dimensional numeric features after assembly; use sparse after one-hot encoding or large vocabularies.

Visual intuition. Picture a length-1000 number line of slots. Dense paints a value in every slot. Sparse leaves slots empty and pins two labeled dots at positions 100 and 200. Takeaway: sparsity stores the pins, not the empty slots.

Real-world & domain. One-hot encoding of high-cardinality categoricals (sku_id, user_segment, geo) produces high-dimensional vectors stored as sparse vectors in Spark ML to avoid memory overflow — the same pattern used in large-scale logistic regression (§15.1.3).

15.3 Core Abstractions: Transformers vs. Estimators

Building Spark ML pipelines requires a sharp operational distinction: Transformers apply fixed rules; Estimators learn parameters from data and then return a Transformer.

15.3.1 Transformer Mechanics and Immutability Guarantees

Key Transformer properties:

• No preliminary full-data scan; rules are deterministic and self-contained.
• Configured with setInputCol() / setInputCols() and optional setOutputCol().
• Executed via .transform(df).

from pyspark.ml.feature import Tokenizer

# Instantiate Transformer
tokenizer = Tokenizer()
tokenizer.setInputCol("description")
tokenizer.setOutputCol("words")

# Execute transformation (appends 'words' column to sales_df)
tokenized_df = tokenizer.transform(sales_df)

15.3.2 Estimator Mechanics and Two-Pass Execution Flow

Min-Max scaling is the canonical example: you cannot scale a single row until you know global \(x_{\min}\) and \(x_{\max}\). Until .fit() finishes, transformation is undefined.

Two-pass architecture (inputs → steps → outputs).

1. First pass — .fit(df): Scan the DataFrame; compute metadata or train parameters; return a fitted Transformer (often named *Model).
2. Second pass — .transform(df): Apply the learned rules to produce the output DataFrame.

from pyspark.ml.feature import MinMaxScaler

# Instantiate Estimator
scaler = MinMaxScaler()
scaler.setInputCol("features")
scaler.setOutputCol("scaledFeatures")

# Pass 1: Fit estimator to compute global min and max statistics across dataset
scaler_model = scaler.fit(df)  # Returns a MinMaxScalerModel (Transformer)

# Pass 2: Apply transformer to scale feature values
scaled_df = scaler_model.transform(df)

Comparison.

When to pick which: if the rule is fully known from configuration alone → Transformer; if the rule depends on data statistics or learned parameters → Estimator.

Visual intuition. Draw a two-box flowchart: DataFrame → fit → Model (stores \(x_{\min}, x_{\max}\)) → transform → Scaled DataFrame. For Tokenizer, collapse to a single arrow: DataFrame → transform → Token DataFrame. Takeaway: Estimators insert an obligatory learn step before transform.

Real-world & domain. Production recommendation and fraud pipelines fit scalers and indexers on training partitions only, package them inside a PipelineModel, and ship that model so scoring clusters apply the same learned mins, maxes, and category maps — never re-fit at serve time on a single batch.

15.4 Feature Transformation and Feature Engineering Toolkit

Spark MLlib ships transformers and estimators for numerical, categorical, text, SQL, and dimensionality-reduction feature work. Each tool is either a one-pass Transformer or a two-pass Estimator (§15.3). Companion texts list feature standardization and related primitives among core MLlib utilities used before classification, regression, and recommendation.

15.4.1 VectorAssembler

from pyspark.ml.feature import VectorAssembler

# Combine separate feature columns into a single 'features' vector
assembler = VectorAssembler()
assembler.setInputCols(["feature_1", "feature_2", "feature_3"])
assembler.setOutputCol("features")

# Apply transformation
assembled_df = assembler.transform(df)

15.4.2 Bucketizer

from pyspark.ml.feature import Bucketizer

# Define explicit bucket split boundaries
splits = [-float("inf"), 0.0, 10.0, 50.0, float("inf")]

bucketizer = Bucketizer()
bucketizer.setSplits(splits)
bucketizer.setInputCol("age")
bucketizer.setOutputCol("age_bucket")

bucketed_df = bucketizer.transform(df)

15.4.3 QuantileDiscretizer

from pyspark.ml.feature import QuantileDiscretizer

# Set up 5 percentile buckets
discretizer = QuantileDiscretizer()
discretizer.setNumBuckets(5)
discretizer.setInputCol("income")
discretizer.setOutputCol("income_quantile")

# Fit estimator to compute quantile boundaries, then transform
discretizer_model = discretizer.fit(df)
quantile_df = discretizer_model.transform(df)

Comparison — Bucketizer vs QuantileDiscretizer.

When to pick which: fixed business rules → Bucketizer; exploratory equal-frequency bins → QuantileDiscretizer.

15.4.4 StandardScaler

Limiting check: if \(\sigma \rightarrow 0\) (constant column), division is undefined — drop or avoid scaling that column. If \(\mu=0\) already and \(\sigma=1\), then \(z=x\).

from pyspark.ml.feature import StandardScaler

scaler = StandardScaler()
scaler.setWithMean(True)
scaler.setWithStd(True)
scaler.setInputCol("features")
scaler.setOutputCol("scaled_features")

scaler_model = scaler.fit(df)
standardized_df = scaler_model.transform(df)

15.4.5 MinMaxScaler

Special case: \(\min_{\mathrm{new}}=0\), \(\max_{\mathrm{new}}=1\) recovers \(x'=(x-x_{\min})/(x_{\max}-x_{\min})\). Boundary check: \(x=x_{\min}\Rightarrow x'=\min_{\mathrm{new}}\); \(x=x_{\max}\Rightarrow x'=\max_{\mathrm{new}}\).

from pyspark.ml.feature import MinMaxScaler

scaler = MinMaxScaler()
scaler.setMin(0.0)
scaler.setMax(1.0)
scaler.setInputCol("features")
scaler.setOutputCol("scaled_features")

scaler_model = scaler.fit(df)
rescaled_df = scaler_model.transform(df)

15.4.6 ElementwiseProduct

from pyspark.ml.feature import ElementwiseProduct
from pyspark.ml.linalg import Vectors

# Define scaling weight vector
scaling_vector = Vectors.dense([10.0, 15.0, 20.0])

transformer = ElementwiseProduct()
transformer.setScalingVector(scaling_vector)
transformer.setInputCol("features")
transformer.setOutputCol("scaled_features")

scaled_df = transformer.transform(df)

15.4.7 Categorical Feature Encoding: StringIndexer and IndexToString

from pyspark.ml.feature import StringIndexer, IndexToString

# Forward encoding: String category to double index
indexer = StringIndexer()
indexer.setInputCol("category_str")
indexer.setOutputCol("category_idx")

indexer_model = indexer.fit(df)
indexed_df = indexer_model.transform(df)

# Reverse decoding: Double index back to string category
converter = IndexToString()
converter.setInputCol("prediction_idx")
converter.setOutputCol("predicted_category_str")

reverted_df = converter.transform(indexed_df)

15.4.8 OneHotEncoder

from pyspark.ml.feature import OneHotEncoder

encoder = OneHotEncoder()
encoder.setInputCol("category_idx")
encoder.setOutputCol("category_vec")

encoder_model = encoder.fit(indexed_df)
encoded_df = encoder_model.transform(indexed_df)

15.4.9 SQLTransformer

from pyspark.ml.feature import SQLTransformer

# Execute custom SQL aggregation/transformation within pipeline
sql_transformer = SQLTransformer()
sql_transformer.setStatement(
    "SELECT *, (quantity * unit_price) AS total_amount FROM __THIS__"
)

transformed_df = sql_transformer.transform(sales_df)

15.4.10 Text Processing: Tokenizer, StopWordsRemover, NGram, and CountVectorizer

Spark MLlib provides an end-to-end text path:

1. Tokenizer (Transformer) — split raw text into lowercase tokens on whitespace.
2. StopWordsRemover (Transformer) — drop low-information words (“the”, “is”, “of”); languages include English, Danish, Dutch, Finnish, Turkish, and others.
3. NGram (Transformer) — emit contiguous \(n\)-gram sequences to capture collocation.
4. CountVectorizer (Estimator) — map token collections to sparse term-frequency vectors over a learned vocabulary.

from pyspark.ml.feature import Tokenizer, StopWordsRemover, NGram

# 1. Tokenize text
tokenizer = Tokenizer(inputCol="text", outputCol="words")
words_df = tokenizer.transform(text_df)

# 2. Remove stop words (English default)
remover = StopWordsRemover(inputCol="words", outputCol="filtered_words")
clean_df = remover.transform(words_df)

# 3. Generate bigrams (n=2)
ngram = NGram(n=2, inputCol="filtered_words", outputCol="bigrams")
ngram_df = ngram.transform(clean_df)

15.4.11 Dimensionality Reduction: PCA (Principal Component Analysis)

from pyspark.ml.feature import PCA

pca = PCA()
pca.setK(2)  # Reduce feature space to top 2 principal components
pca.setInputCol("features")
pca.setOutputCol("pca_features")

pca_model = pca.fit(df)
pca_df = pca_model.transform(df)

Visual intuition. Scatter points in \(\mathbb{R}^{2}\) elongated along a diagonal: the first principal component is that diagonal axis; the second is orthogonal and shorter. Projecting to \(k=1\) collapses onto the long axis. Takeaway: PCA keeps the directions where data spreads most.

Real-world & domain. Retail and ad pipelines chain SQLTransformer (derived amounts), StringIndexer + OneHotEncoder (store and segment), StandardScaler or MinMaxScaler, optional PCA, then a classifier — all inside one Pipeline artifact for identical train and serve transforms. Multilingual document analytics use StopWordsRemover language packs before CountVectorizer-based models.

15.5 Distributed Model Training, Hyperparameter Tuning, and Evaluation

Once features are assembled, training instantiates ML Estimators, fits them on training partitions, predicts on test partitions, and evaluates with metric objects. This section uses the procedural spine: purpose, inputs/outputs, steps, concrete traces, cost notes, and when to use alternatives.

15.5.1 End-to-End Supervised Regression Workflow

Inputs & Outputs.

Steps.

1. Create SparkSession and read data.
2. Assemble feature columns; select features + label.
3. Split with a fixed seed for reproducibility.
4. Instantiate LinearRegression; optionally explainParams().
5. .fit(train_data) → model; read coefficients, intercept.
6. .transform(test_data) → predictions.
7. RegressionEvaluator.evaluate(...) → metric.

from pyspark.ml.feature import VectorAssembler
from pyspark.ml.regression import LinearRegression
from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.sql import SparkSession

# Initialize Spark Session
spark = SparkSession.builder.appName("SparkMLExample").getOrCreate()

# 1. Read input dataset
data = spark.read.csv("study_grades.csv", header=True, inferSchema=True)

# 2. Assemble feature columns into single 'features' vector
assembler = VectorAssembler(inputCols=["study_time"], outputCol="features")
final_data = assembler.transform(data).select("features", "grades")

# 3. Split dataset into 70% training and 30% testing sets with seed
train_data, test_data = final_data.randomSplit([0.7, 0.3], seed=42)

# 4. Instantiate Linear Regression Estimator
linreg = LinearRegression(featuresCol="features", labelCol="grades")

# Inspect algorithm hyperparameter options
print(linreg.explainParams())

# 5. Fit Estimator on training dataset to produce trained LinearRegressionModel
model = linreg.fit(train_data)

# Extract learned model parameters (weight coefficient and intercept)
print(f"Model Coefficient: {model.coefficients}")
print(f"Model Intercept: {model.intercept}")

# 6. Generate predictions and evaluate on held-out test dataset
predictions = model.transform(test_data)
predictions.select("features", "grades", "prediction").show(5)

# 7. Evaluate model performance using RegressionEvaluator
evaluator = RegressionEvaluator(
    labelCol="grades", predictionCol="prediction", metricName="r2"
)
r2_score = evaluator.evaluate(predictions)
print(f"Test R-Squared Score: {r2_score}")

Complexity & cost. Linear regression on Spark scales with rows × features × iterations of the underlying optimizer; sparse high-\(d\) problems lean on sparse vector kernels (§15.1.3). Shuffle cost appears mainly at the train/test split and during metric aggregation.

15.5.2 Automated Hyperparameter Tuning with ParamGridBuilder

ParamGridBuilder builds the Cartesian product of parameter lists. Example: 3 values of elasticNetParam × 2 values of regParam ⇒ \[ 3 \times 2 = 6 \] model configurations. CrossValidator trains each configuration on numFolds folds (here 3), scores with an Evaluator, and keeps bestModel.

from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
from pyspark.ml.evaluation import RegressionEvaluator

# Construct parameter grid
param_grid = (
    ParamGridBuilder()
    .addGrid(linreg.elasticNetParam, [0.0, 0.5, 1.0])
    .addGrid(linreg.regParam, [0.01, 0.1])
    .build()
)

# Set up cross-validation pipeline
cross_val = CrossValidator(
    estimator=linreg,
    estimatorParamMaps=param_grid,
    evaluator=RegressionEvaluator(labelCol="grades", metricName="rmse"),
    numFolds=3,
)

# Fit cross-validator to execute 6 model variant trials
cv_model = cross_val.fit(train_data)
best_model = cv_model.bestModel

15.5.3 Distributed Classification and Binary Evaluators

BinaryClassificationEvaluator measures ranking quality via Area Under the ROC Curve (AUC). ROC plots true-positive rate (vertical) against false-positive rate (horizontal) as the decision threshold moves; AUC summarizes that curve into one number in \([0,1]\) (0.5 ≈ random, 1.0 ≈ perfect separation).

from pyspark.ml.classification import RandomForestClassifier
from pyspark.ml.evaluation import BinaryClassificationEvaluator

# Instantiate Random Forest Classifier
rf = RandomForestClassifier(
    featuresCol="features", labelCol="label", numTrees=100
)

# Train model on training data
rf_model = rf.fit(train_data)

# Generate predictions on test data
predictions = rf_model.transform(test_data)

# Evaluate model using Area Under ROC Curve (AUC)
evaluator = BinaryClassificationEvaluator(
    labelCol="label", metricName="areaUnderROC"
)
auc_score = evaluator.evaluate(predictions)
print(f"Area Under ROC (AUC): {auc_score}")

Visual intuition. ROC curve: x-axis = false-positive rate \(\in[0,1]\), y-axis = true-positive rate \(\in[0,1]\). A curve bowed toward the top-left is better; the area under that curve is AUC. Diagonal line = random classifier.

Real-world & domain. Study-time regression demos the full Estimator → Model → Evaluator path used in tutoring analytics; the same pattern scores credit-risk forests with AUC and tunes elastic-net logistic models on click logs with ParamGridBuilder across a cluster.

Exam Guidance Summary

• Regular Examination Scope: Topics in Lecture 15 (Spark MLlib, feature engineering transformers/estimators, pipelines, and distributed model tuning) will NOT appear on the regular Saturday examination.
• Makeup Examination Scope: Lecture 15 material WILL be in syllabus for makeup-exam students. Questions stay within concepts, code patterns, and abstractions presented in class (Transformer vs Estimator, vector formats, named feature tools, ParamGrid/CrossValidator, evaluators).
• Cassandra Module Note: Apache Cassandra storage integration (~45 minutes) was postponed for time. An optional supplementary session may be scheduled after examinations if students request it.

Key Industry Applications

• Large-Scale Distributed Training: Logistic regression feature spaces up to about \(10^{7}\) features and Random Forests up to about \(10^{4}\) features across commodity clusters — clickstream, ads, and assortment models stay inside Spark’s DataFrame runtime.
• Multi-Language Text Analytics: Enterprise stop-word filtering over multilingual corpora (English, Danish, Dutch, Finnish, Turkish, …) with StopWordsRemover before vectorization and classification.
• SQL Pipeline Integration: Derived business metrics (e.g., quantity * unit_price) computed inside ML pipelines via SQLTransformer and the __THIS__ binding — one artifact for ETL-like features and model scoring.
• Pipeline Standardization: Package feature extraction, model fitting, and evaluation into reproducible Pipeline / PipelineModel deployment units so train-time encodings and scalers match serve-time transforms.