MLlib - Spark's Machine Learning Library

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

# Data must have "features" (Vector) and "label" columns
lr = LogisticRegression(
    maxIter=100,
    regParam=0.01,
    elasticNetParam=0.8,
    featuresCol="features",
    labelCol="label"
)

# Train
model = lr.fit(train_df)

# Predict
predictions = model.transform(test_df)

# Evaluate
evaluator = BinaryClassificationEvaluator(metricName="areaUnderROC")
auc = evaluator.evaluate(predictions)
print(f"AUC-ROC: {auc:.4f}")

# Model coefficients
print(f"Coefficients: {model.coefficients}")
print(f"Intercept: {model.intercept}")

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

rf = RandomForestClassifier(
    numTrees=100,
    maxDepth=10,
    featuresCol="features",
    labelCol="label",
    seed=42
)

model = rf.fit(train_df)
predictions = model.transform(test_df)

evaluator = MulticlassClassificationEvaluator(metricName="accuracy")
accuracy = evaluator.evaluate(predictions)
print(f"Accuracy: {accuracy:.4f}")

# Feature importance
print(f"Feature importances: {model.featureImportances}")

from pyspark.ml.regression import GBTRegressor
from pyspark.ml.evaluation import RegressionEvaluator

gbt = GBTRegressor(
    maxIter=100,
    maxDepth=5,
    stepSize=0.1,
    featuresCol="features",
    labelCol="price"
)

model = gbt.fit(train_df)
predictions = model.transform(test_df)

evaluator = RegressionEvaluator(
    labelCol="price",
    predictionCol="prediction",
    metricName="rmse"
)
rmse = evaluator.evaluate(predictions)
print(f"RMSE: {rmse:.4f}")

from pyspark.ml.clustering import KMeans
from pyspark.ml.evaluation import ClusteringEvaluator

kmeans = KMeans(k=5, seed=42, featuresCol="features")
model = kmeans.fit(df)

predictions = model.transform(df)

evaluator = ClusteringEvaluator()
silhouette = evaluator.evaluate(predictions)
print(f"Silhouette score: {silhouette:.4f}")

# Cluster centers
centers = model.clusterCenters()
for i, center in enumerate(centers):
    print(f"Cluster {i}: {center}")

from pyspark.ml.recommendation import ALS

als = ALS(
    maxIter=10,
    regParam=0.1,
    rank=10,
    userCol="userId",
    itemCol="movieId",
    ratingCol="rating",
    coldStartStrategy="drop"
)

model = als.fit(train_df)

# Predict ratings
predictions = model.transform(test_df)

# Recommend top 10 movies for each user
user_recs = model.recommendForAllUsers(10)
user_recs.show(5, truncate=False)