Spark MLlib 101: Kickstarting Your Machine Learning Journey#

2.1 Spark Architecture#

At a high level, Spark applications run on a cluster in a distributed manner. The integral concepts include:

• Driver: The main process running the user code. It transforms user code into tasks and distributes them to the cluster executors.
• Executor: The processes that run on the cluster nodes (workers). They are assigned tasks by the driver.
• Cluster Manager: Manages resources among all applications running on the cluster. This can be Spark’s own standalone cluster manager, or external ones like YARN or Mesos.

2.2 RDD vs. DataFrame vs. DataSet#

Spark has evolved to provide multiple abstractions for data processing:

As of Spark 2.x and above, DataFrames have become the primary API for machine learning as they provide a simpler, more optimized abstraction. MLlib also supports the older RDD-based APIs, but new development is focused on DataFrame-based ML pipelines.

4.1 Setting Up Your Environment#

Spark can be run locally on your computer, or on a cluster. The easiest way for beginners:

1. Download and install Apache Spark.
2. Make sure Java (and optionally Scala or Python) is installed.
3. Use the built-in spark-shell or a Jupyter notebook with pyspark.

1
pip install pyspark

1
pyspark

4.2 Basic Data Processing Workflow#

1. Launch SparkSession: The SparkSession object is the entry point for all Spark operations.
2. Load Data: Read data from various sources (CSV, Parquet, JSON, etc.).
3. Transformations: Clean, filter, and join your data.
4. Machine Learning: Apply MLlib transformations, create pipelines, and train models.
5. Evaluations: Evaluate model performance using built-in metrics or custom logic.

4.3 Loading and Parsing Data#

Let’s say you have a CSV file containing housing data. A simple way to load this data into a Spark DataFrame in PySpark:

1
from pyspark.sql import SparkSession
2

3
spark = SparkSession.builder \
4
    .appName("HousingDataAnalysis") \
5
    .getOrCreate()
6

7
# Load CSV data
8
df = spark.read.csv("housing_data.csv", header=True, inferSchema=True)
9

10
# Inspect schema
11
df.printSchema()
12

13
# Show a few rows
14
df.show(5)

Here, Spark automatically infers the schema from the CSV file. You can also manually specify the schema if needed.

5.1 Classification#

Classification aims to predict a discrete label (e.g., “spam” vs. “not spam”). Popular algorithms in MLlib include:

• Logistic Regression
• Decision Tree
• Random Forest
• Gradient-Boosted Trees
• Naive Bayes
• Support Vector Machines (SVM)

Example: Logistic Regression in Spark DataFrames

1
from pyspark.ml.classification import LogisticRegression
2
from pyspark.ml.feature import VectorAssembler, StringIndexer
3

4
# Example dataset with 'text' and 'label' columns
5
# We'll just illustrate the pipeline creation step:
6

7
# Convert text label into numerical
8
indexer = StringIndexer(inputCol="label", outputCol="labelIndexed")
9

10
# Assemble multiple features into a single vector
11
assembler = VectorAssembler(
12
    inputCols=["feature1", "feature2", "feature3"],
13
    outputCol="features"
14
)
15

16
# Initialize the Logistic Regression model
17
lr = LogisticRegression(featuresCol="features", labelCol="labelIndexed")
18

19
# Build the pipeline
20
from pyspark.ml import Pipeline
21

22
pipeline = Pipeline(stages=[indexer, assembler, lr])
23
model = pipeline.fit(trainingData)
24

25
# Predictions
26
predictions = model.transform(testData)
27
predictions.select("features", "labelIndexed", "prediction").show(5)

5.2 Regression#

Regression models predict a continuous value. Common MLlib regressors include:

• Linear Regression
• Decision Tree Regressor
• Random Forest Regressor
• Gradient-Boosted Tree Regressor

Example: Linear Regression

1
from pyspark.ml.regression import LinearRegression
2

3
assembler = VectorAssembler(inputCols=["sqft", "bedrooms", "bathrooms"], outputCol="features")
4
lr = LinearRegression(featuresCol="features", labelCol="price")
5

6
pipeline = Pipeline(stages=[assembler, lr])
7
model = pipeline.fit(trainingData)
8

9
# Evaluate
10
predictions = model.transform(testData)
11
predictions.select("price", "prediction").show(5)

5.3 Clustering#

Clustering groups data points without existing labels. Spark MLlib supports:

• K-means
• Gaussian Mixture Model (GMM)
• Bisecting k-means
• Latent Dirichlet Allocation (Topic Modeling)

Example: K-means

1
from pyspark.ml.clustering import KMeans
2

3
assembler = VectorAssembler(inputCols=["col1", "col2"], outputCol="features")
4
kmeans = KMeans(k=3, seed=1)
5

6
pipeline = Pipeline(stages=[assembler, kmeans])
7
model = pipeline.fit(df)
8

9
centers = model.stages[-1].clusterCenters()
10
print("Cluster Centers: ", centers)

5.4 Collaborative Filtering#

Used for recommendation systems (e.g., movie recommendations). MLlib provides an implementation of Alternating Least Squares (ALS) for building collaborative filtering models.

1
from pyspark.ml.recommendation import ALS
2

3
als = ALS(userCol="userId", itemCol="movieId", ratingCol="rating", coldStartStrategy="drop")
4
model = als.fit(trainingData)
5

6
# Generate top 10 movie recommendations for each user
7
userRecs = model.recommendForAllUsers(10)
8
userRecs.show()

5.5 Dimensionality Reduction#

High-dimensional data can lead to slow processing and overfitting. MLlib includes:

• PCA (Principal Component Analysis)
• SVD (Singular Value Decomposition)

1
from pyspark.ml.feature import PCA
2

3
assembler = VectorAssembler(inputCols=["col1", "col2", "col3", "col4"], outputCol="features")
4
pca = PCA(k=2, inputCol="features", outputCol="pcaFeatures")
5

6
pipeline = Pipeline(stages=[assembler, pca])
7
model = pipeline.fit(df)
8

9
result = model.transform(df)
10
result.select("pcaFeatures").show(5, truncate=False)

6.1 Feature Transformers#

Some built-in transformers:

• StringIndexer: Converts categorical text labels into numerical labels.
• OneHotEncoder: Converts categorical features into a one-hot (dummy) vector.
• VectorAssembler: Combines multiple feature columns into a single vector.
• Normalizer: Normalize feature vectors to unit norms.
• MinMaxScaler: Scales data to a specified range [min, max].
• Binarizer: Threshold-based binarization of numeric feature data.

6.2 Vector Assemblers, Indexers, and Encoders#

A common workflow in MLlib pipelines:

1. Use StringIndexer to encode categorical columns into integer indices.
2. Use OneHotEncoder (or newer OneHotEncoderEstimator) to convert indexed columns into binary vectors.
3. Use VectorAssembler to merge all feature columns into a single features vector.

1
from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler
2

3
# Convert category text to numeric index
4
indexer = StringIndexer(inputCol="category", outputCol="categoryIndex")
5

6
# Create one-hot vectors
7
encoder = OneHotEncoder(inputCol="categoryIndex", outputCol="categoryVec")
8

9
# Assemble final features
10
assembler = VectorAssembler(inputCols=["categoryVec", "numericalColumn1"], outputCol="features")
11

12
pipeline = Pipeline(stages=[indexer, encoder, assembler])
13
final_model = pipeline.fit(df)
14
transformed = final_model.transform(df)

7.1 Concept of Pipelines#

MLlib’s Pipeline API treats a machine learning workflow as a sequence of stages:

1. Transformer: A function to transform one DataFrame into another (e.g., StringIndexer).
2. Estimator: Learns from data and produces a Transformer (e.g., a learning algorithm like LogisticRegression).

Data flow is orchestrated from raw data transformations to model training. Once a pipeline is fit on training data, you can apply the resulting PipelineModel to new data for predictions.

7.2 Evaluation and Model Tuning#

Spark MLlib provides evaluation metrics like:

• BinaryClassificationEvaluator: For binary classification tasks, measuring metrics like area under ROC or precision–recall.
• MulticlassClassificationEvaluator: For multi-class metrics.
• RegressionEvaluator: For regression metrics like RMSE, MAE, R².

You typically combine these evaluators with hyperparameter tuning in order to find the best model parameters.

8.1 Cross Validation and Hyperparameter Tuning#

CrossValidator in MLlib automates hyperparameter search:

1
from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
2
from pyspark.ml.evaluation import RegressionEvaluator
3

4
lr = LinearRegression(featuresCol="features", labelCol="price")
5

6
paramGrid = ParamGridBuilder() \
7
    .addGrid(lr.regParam, [0.1, 0.01]) \
8
    .addGrid(lr.elasticNetParam, [0.0, 0.5, 1.0]) \
9
    .build()
10

11
evaluator = RegressionEvaluator(labelCol="price", metricName="rmse")
12

13
crossval = CrossValidator(estimator=lr,
14
                          estimatorParamMaps=paramGrid,
15
                          evaluator=evaluator,
16
                          numFolds=5)
17

18
cvModel = crossval.fit(trainingData)
19
bestModel = cvModel.bestModel

In this example, Spark will train multiple Linear Regression models with different regularization and elasticity parameters. It will pick the model that yields the best RMSE after 5-fold cross-validation.

8.2 MLflow and Model Tracking#

While Spark MLlib streamlines training, you often need to track experiments:

• MLflow: An open-source platform for managing the ML lifecycle.
• You can log Spark model parameters and metrics to MLflow, version your models, and serve them in production easily.

8.3 Deploying Spark MLlib Models#

After you’ve trained a model, you might want to:

1. Save the model and pipeline:

   1
   model.save("myModelPath")
   2
   pipeline.write().overwrite().save("myPipelinePath")
   

2. Load it in another Spark job:

   1
   from pyspark.ml import PipelineModel
   2
   loadedPipeline = PipelineModel.load("myPipelinePath")
   

3. Serve your model as a service via Spark Streaming or batch jobs that score incoming data.

8.4 Streaming and Real-Time ML#

Spark Streaming (or Structured Streaming) allows for real-time data processing. You can integrate your MLlib models into a streaming job:

• Fetch streaming data from sources like Kafka.
• Apply the same transformations and pipeline used in batch mode.
• Continuously update or apply a pre-trained model to new data.

10.1 Data Ingestion#

1
spark = SparkSession.builder.appName("LoanApp").getOrCreate()
2
df = spark.read.csv("loan_applications.csv", header=True, inferSchema=True)
3
df.printSchema()
4
df.show(5)

10.2 Data Cleaning and Feature Engineering#

1. Handle missing values:

   1
   df = df.na.fill({"employmentLength": 0})
   

2. Encoding categorical features:

   1
   from pyspark.ml.feature import StringIndexer
   2
   
   3
   purposeIndexer = StringIndexer(inputCol="loanPurpose", outputCol="loanPurposeIndex")
   4
   stateIndexer = StringIndexer(inputCol="state", outputCol="stateIndex")
   

3. Assemble final features:

   1
   from pyspark.ml.feature import VectorAssembler
   2
   
   3
   assembler = VectorAssembler(
   4
       inputCols=["creditScore", "income", "loanAmount", "employmentLength",
   5
                  "loanPurposeIndex", "stateIndex"],
   6
       outputCol="features"
   7
   )
   

10.3 Model Training and Evaluation#

1
from pyspark.ml.classification import RandomForestClassifier
2
from pyspark.ml import Pipeline
3

4
rf = RandomForestClassifier(labelCol="defaulted", featuresCol="features")
5

6
pipeline = Pipeline(stages=[purposeIndexer, stateIndexer, assembler, rf])
7
trainData, testData = df.randomSplit([0.8, 0.2], seed=42)
8

9
model = pipeline.fit(trainData)
10
predictions = model.transform(testData)
11

12
from pyspark.ml.evaluation import BinaryClassificationEvaluator
13

14
evaluator = BinaryClassificationEvaluator(
15
    labelCol="defaulted", rawPredictionCol="rawPrediction", metricName="areaUnderROC"
16
)
17
auc = evaluator.evaluate(predictions)
18
print("AUC on Test Data = ", auc)

10.4 Deployment Strategy#

• Batch scoring: Schedule a Spark job daily to load new loan applications, apply the pipeline model, and store predictions.
• Real-time scoring: Load the same pipeline in a Structured Streaming job, read real-time data from an event source like Kafka, transform, and score instantly.