Analyzing and predicting student academic performance allows educational institutions to identify at-risk students and provide timely interventions. This technical guide explores how to build a predictive pipeline using PySpark and Scikit-learn, focusing on data preprocessing, feature correlation, and regression modeling.

Data Analysis and Scikit-Learn Baseline

Before implementing a distributed solution with PySpark, it is common practice to perform initial exploratory data enalysis (EDA) and establish a baseline model using Pandas and Scikit-learn. The primary goal is to identify which factors—such as study time, parental education, and previous failures—most significantly influence the final grade (G3).

import pandas as pd
import numpy as np
import pickle
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_absolute_error, r2_score, mean_squared_error

# Load the student performance dataset
df_source = pd.read_csv('../data/student-mat.csv')

# Calculate correlation to identify key performance drivers
correlations = df_source.corr(numeric_only=True)['G3'].abs().sort_values(ascending=False)
print("Top correlated features:\n", correlations.head(10))

# Feature selection based on impact analysis
selected_features = ['G3', 'failures', 'Medu', 'age', 'Fedu', 'goout', 'traveltime', 'romantic', 'higher']
filtered_data = df_source[selected_features]

# Prepare data: One-hot encode categorical variables
processed_data = pd.get_dummies(filtered_data, drop_first=True)
X = processed_data.drop('G3', axis=1)
y = processed_data['G3']

# Split the dataset for training and validation
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.15, random_state=42)

# Initialize and train the Linear Regression model
regressor = LinearRegression()
regressor.fit(X_train, y_train)

# Evaluate the model
y_val_pred = regressor.predict(X_test)
print(f"R-Squared: {round(r2_score(y_test, y_val_pred), 2)}")
print(f"MAE: {round(mean_absolute_error(y_test, y_val_pred), 2)}")

# Serialize the model for production use
with open('../models/performance_model.pkl', 'wb') as f:
    pickle.dump(regressor, f)

Building a Distributed Pipeline with PySpark

To handle larger datasets or integrate with Big Data ecosystems, we transition to PySpark's MLlib. The process involves creating a SparkSession, handling categorical strings via indexing, and assembling features in to a vector format required by Spark's machine learning algorithms.

from pyspark.sql import SparkSession
from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler
from pyspark.ml.regression import LinearRegression, LinearRegressionModel
from pyspark.ml import Pipeline

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

# Load dataset
raw_spark_df = spark.read.csv('student-mat.csv', header=True, inferSchema=True)

# Define preprocessing stages for categorical columns
categorical_cols = ["school", "sex", "romantic", "higher"]
indexers = [StringIndexer(inputCol=col, outputCol=f"{col}_idx") for col in categorical_cols]
encoders = [OneHotEncoder(inputCol=f"{col}_idx", outputCol=f"{col}_vec") for col in categorical_cols]

# Combine numerical and processed categorical features
assembler_inputs = [f"{col}_vec" for col in categorical_cols] + ["failures", "Medu", "age", "Fedu"]
feature_assembler = VectorAssembler(inputCols=assembler_inputs, outputCol="features")

# Assemble the ML Pipeline
ml_pipeline = Pipeline(stages=indexers + encoders + [feature_assembler])
pipeline_model = ml_pipeline.fit(raw_spark_df)
final_df = pipeline_model.transform(raw_spark_df).select("features", "G3")

# Train/Test Split
training_set, testing_set = final_df.randomSplit([0.85, 0.15], seed=42)

# Initialize Linear Regression
lr_algo = LinearRegression(featuresCol='features', labelCol='G3')
trained_lr_model = lr_algo.fit(training_set)

# Evaluate performance on test data
summary = trained_lr_model.evaluate(testing_set)
print(f"Test RMSE: {summary.rootMeanSquaredError}")
print(f"Test R2 Score: {summary.r2}")

Model Persistence and Prediction Visualization

Once the model achieves satisfactoyr metrics, it is saved for future inference. Visualizing the variance between predicted values and actual grades helps in understanding the model's error distribution across different score ranges.

import matplotlib.pyplot as plt
import seaborn as sns

# Generate predictions for visualization
predictions_df = trained_lr_model.transform(testing_set)
pdf = predictions_df.select("G3", "prediction").toPandas()

# Plotting Actual vs Predicted values
plt.figure(figsize=(10, 6))
sns.lineplot(data=pdf['prediction'], label='Predicted Grade', color='blue')
sns.lineplot(data=pdf['G3'], label='Actual Grade', color='orange', alpha=0.6)
plt.title("Performance Comparison: Predicted vs Actual")
plt.xlabel("Student Sample Index")
plt.ylabel("Final Grade (G3)")
plt.legend()
plt.show()

# Save the Spark ML model
trained_lr_model.write().overwrite().save("./models/pyspark_grade_regressor")

# Loading the model for a single inference
loaded_model = LinearRegressionModel.load("./models/pyspark_grade_regressor")
sample_data = testing_set.limit(1)
result = loaded_model.transform(sample_data)
result.select("features", "G3", "prediction").show()