Implement a PySpark solution to predict customer churn using a large dataset of user activity and transactions.

Instruction: Outline the data preparation, feature engineering, and model training steps in your approach.

Context: Candidates should demonstrate their skills in machine learning pipeline construction, from raw data to predictive modeling, using PySpark's MLlib or integrating with other machine learning frameworks.

Updated

Define churn, an observation cutoff and a future label horizon. Build one eligible row per customer/cutoff using only activity and transactions available at that cutoff; wait for the label horizon before training. Respect customer/time separation and keep the final assessment outside model/threshold selection.

This small API demonstration assumes an existing spark session and already prepared numeric features. The train/test labels and features are synthetic; their metrics are not evidence of a useful churn model:

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

train = spark.createDataFrame([
    (1, 1.0, 10.0, 0.0), (2, 2.0, 8.0, 0.0),
    (3, 3.0, 9.0, 0.0), (4, None, 7.0, 0.0),
    (5, 20.0, 1.0, 1.0), (6, 25.0, 0.0, 1.0),
    (7, 18.0, 2.0, 1.0), (8, 30.0, 1.0, 1.0)
], "id int, recency double, activity double, label double")
test = spark.createDataFrame([
    (9, 2.0, 9.0, 0.0), (10, 4.0, 7.0, 0.0),
    (11, 22.0, 1.0, 1.0), (12, 28.0, 0.0, 1.0)
], train.schema)
pipeline = Pipeline(stages=[
    Imputer(inputCols=["recency"], outputCols=["recency_filled"]),
    VectorAssembler(inputCols=["recency_filled", "activity"], outputCol="features"),
    LogisticRegression(maxIter=30, regParam=0.1)
])
model = pipeline.fit(train)
scored = model.transform(test)
pr_auc = BinaryClassificationEvaluator(metricName="areaUnderPR").evaluate(scored)
assert scored.count() == 4 and 0 <= pr_auc <= 1
assert scored.select("prediction").distinct().count() <= 2

Fit all learned preprocessing inside training folds when tuning; this example fits only the supplied training set. In a real pipeline, define missing/invalid-feature policies, prevent join multiplication and reconcile eligible versus excluded customers. Compare a prevalence/rule baseline with logistic regression before adding more complex estimators.

The binary evaluator supplies ROC/PR area metrics, not arbitrary threshold precision/recall/F1. Compute those from a declared positive class and threshold/confusion counts, or configure the appropriate multiclass evaluator carefully; weighted aggregate metrics are not the same as churn-class recall. Also assess calibration, cost and subgroup coverage.

Publish versioned features/model metadata, use the same transformations at scoring, monitor drift and delayed outcomes, and test model loading/recovery. A high churn score does not establish which retention intervention will work; that requires separate evidence.

ML pipelines, Binary evaluator

Related Questions